diff --git a/.gitea/workflows/whisper-image.yml b/.gitea/workflows/whisper-image.yml new file mode 100644 index 0000000..e0184da --- /dev/null +++ b/.gitea/workflows/whisper-image.yml @@ -0,0 +1,83 @@ +name: Whisper Image + +# Nur bei Aenderungen am Abschreib-Image (Spec T7 11.1): das Image ist rund +# 1,2 GB gross (Modell + Python-Abhaengigkeiten) und aendert sich selten. +# Tag sha-; das Overlay setzt danach den Digest. +on: + push: + branches: [main] + paths: ['ApiCore-Whisper/**', '.gitea/workflows/whisper-image.yml'] + workflow_dispatch: + +jobs: + build-and-push: + runs-on: docker-builder + env: + # Nur fuer Doku/OCI-Labels, siehe REGISTRY_PUSH fuer den eigentlichen Push. + REGISTRY: git.encry.de + # Cluster-interner Service-Name statt git.encry.de: Registry-Blob- + # Uploads (PUT) vertragen sich nicht mit dem Cloudflare-Proxy vor + # git.encry.de ("413 Payload Too Large" unabhaengig von der Groesse). + # Der Runner-Pod erreicht den Gitea-Service direkt ueber Kubernetes-DNS. + # Push und Pull landen im selben Registry-Pfad (Owner/Repo:Tag), der + # Hostname ist nur die Verbindung, daher bleibt der Pull in den + # k8s-Manifesten unveraendert ueber git.encry.de. + REGISTRY_PUSH: gitea:80 + steps: + - name: Checkout + uses: actions/checkout@11d5960a326750d5838078e36cf38b85af677262 # v4 + + - name: Image-Namen in Kleinbuchstaben ermitteln + # OCI-Registries erlauben nur Kleinbuchstaben im Repository-Pfad; + # der Gitea-Owner-Name behaelt Gross-/Kleinschreibung bei. + id: image + run: echo "name=$(echo '${{ gitea.repository_owner }}/apicore-whisper' | tr '[:upper:]' '[:lower:]')" >> "$GITEA_OUTPUT" + + - name: Docker Buildx einrichten + uses: docker/setup-buildx-action@8d2750c68a42422c14e847fe6c8ac0403b4cbd6f # v3 + with: + # REGISTRY_PUSH ist Klartext-HTTP (Gitea-Service intern ohne TLS) + # statt HTTPS — BuildKit lehnt das ohne diese Freigabe ab. + config-inline: | + [registry."gitea:80"] + http = true + + - name: Bei der Gitea-Container-Registry anmelden + uses: docker/login-action@c94ce9fb468520275223c153574b00df6fe4bcc9 # v3 + with: + registry: ${{ env.REGISTRY_PUSH }} + username: ${{ gitea.actor }} + # GITEA_TOKEN hat kein Package-Write-Recht — separates Personal + # Access Token (Scope "package") als Secret REGISTRY_TOKEN anlegen. + password: ${{ secrets.REGISTRY_TOKEN }} + + - name: Metadaten (Tags/Labels) ermitteln + id: meta + uses: docker/metadata-action@c299e40c65443455700f0fdfc63efafe5b349051 # v5 + with: + images: ${{ env.REGISTRY_PUSH }}/${{ steps.image.outputs.name }} + tags: | + type=sha,format=short + + - name: Image bauen und pushen + uses: docker/build-push-action@10e90e3645eae34f1e60eeb005ba3a3d33f178e8 # v6 + with: + context: ./ApiCore-Whisper + push: true + tags: ${{ steps.meta.outputs.tags }} + labels: ${{ steps.meta.outputs.labels }} + cache-from: type=gha,scope=apicore-whisper + cache-to: type=gha,scope=apicore-whisper,mode=max + + - name: Schwachstellen-Scan (nur wenn trivy im Runner vorhanden ist) + # Hinweisend (exit-code 0): die Ausgabe gehoert ins Runbook, ein Fund + # blockiert den Push nicht, weil das Image dann schon in der Registry + # liegt. Der Build selbst ist die harte Sperre (Hashes, Modell-SHA, + # pytest in der Test-Stufe). + run: | + if command -v trivy >/dev/null 2>&1; then + first_tag="$(echo '${{ steps.meta.outputs.tags }}' | head -n1)" + trivy image --insecure --severity HIGH,CRITICAL --exit-code 0 "$first_tag" + else + echo "trivy ist im Runner nicht installiert, Scan uebersprungen" + fi diff --git a/ApiCore-Whisper/.dockerignore b/ApiCore-Whisper/.dockerignore new file mode 100644 index 0000000..e9b5cab --- /dev/null +++ b/ApiCore-Whisper/.dockerignore @@ -0,0 +1,5 @@ +__pycache__ +*.pyc +.pytest_cache +.venv +venv diff --git a/ApiCore-Whisper/.gitignore b/ApiCore-Whisper/.gitignore new file mode 100644 index 0000000..234810a --- /dev/null +++ b/ApiCore-Whisper/.gitignore @@ -0,0 +1,3 @@ +__pycache__/ +.pytest_cache/ +.venv/ diff --git a/ApiCore-Whisper/Dockerfile b/ApiCore-Whisper/Dockerfile new file mode 100644 index 0000000..4f1dfec --- /dev/null +++ b/ApiCore-Whisper/Dockerfile @@ -0,0 +1,58 @@ +# syntax=docker/dockerfile:1 + +# Abschreib-Image fuer den CronJob apicore-job-voice-transcribe (Spec T7 6.3, +# 11.1). Zur Laufzeit ohne Netz, ohne Schluessel, ohne Datenbank: das Image +# enthaelt Python, faster-whisper, PyAV und das Modell `small`, sonst nichts. +# +# Base-Image per Digest gepinnt (python:3.12-slim, Multi-Arch-Index vom +# 2026-10-16, Python 3.12.15). Ein Update ist ein bewusster Commit mit neuem +# Digest, ermittelt per +# docker buildx imagetools inspect python:3.12-slim +ARG BASE_IMAGE=python:3.12-slim@sha256:a6e34c598f2467ed0e9a8d349809fcd8b5c603269512df273a0bb1784edc11b1 + +# --- Modell: feste Revision, jede Datei gegen ihre SHA-256 geprueft --------- +FROM ${BASE_IMAGE} AS model +WORKDIR /build +COPY fetch_model.py ./ +RUN python fetch_model.py /models/small + +# --- Abhaengigkeiten: nur Wheels, nur mit Hash (requirements.txt) ----------- +FROM ${BASE_IMAGE} AS deps +ENV PIP_DISABLE_PIP_VERSION_CHECK=1 PIP_NO_CACHE_DIR=1 +RUN python -m venv /opt/venv +COPY requirements.txt /tmp/requirements.txt +RUN /opt/venv/bin/pip install --require-hashes --only-binary=:all: -r /tmp/requirements.txt + +# --- Tests: ohne bestandene Tests entsteht kein Image ----------------------- +FROM deps AS test +ENV PIP_DISABLE_PIP_VERSION_CHECK=1 PIP_NO_CACHE_DIR=1 PYTHONDONTWRITEBYTECODE=1 +COPY requirements.txt requirements-test.txt /tmp/ +RUN /opt/venv/bin/pip install --require-hashes --only-binary=:all: -r /tmp/requirements.txt -r /tmp/requirements-test.txt +WORKDIR /app +COPY transcribe.py fetch_model.py vocabulary.txt ./ +COPY tests ./tests +RUN /opt/venv/bin/python -m pytest -q -p no:cacheprovider tests && touch /app/.tests-passed + +# --- Laufzeit --------------------------------------------------------------- +# Kein PYTHONUNBUFFERED/PYTHONDONTWRITEBYTECODE: `python -I` ignoriert PYTHON*- +# Variablen; die Ausgabe sind wenige Zaehlerzeilen am Ende des Laufs. +FROM ${BASE_IMAGE} AS runtime +# uid/gid 1000 wie die Backend-Container: fetch, whisper und ingest teilen sich +# das Arbeitsverzeichnis (Modus 0700/0600) und brauchen dieselbe Kennung. +RUN groupadd --gid 1000 whisper \ + && useradd --uid 1000 --gid 1000 --no-create-home --home-dir /tmp --shell /usr/sbin/nologin whisper +ENV PATH=/opt/venv/bin:$PATH \ + HOME=/tmp \ + HF_HOME=/tmp/hf \ + HF_HUB_OFFLINE=1 \ + TRANSFORMERS_OFFLINE=1 \ + WHISPER_MODEL_DIR=/models/small +COPY --from=deps /opt/venv /opt/venv +COPY --from=model /models /models +WORKDIR /app +COPY transcribe.py vocabulary.txt ./ +# Haengt die Laufzeit an die Test-Stufe (BuildKit baut sonst nur, was gebraucht +# wird), ohne dass etwas davon im Image landet. +RUN --mount=type=bind,from=test,source=/app/.tests-passed,target=/mnt/tests-passed true +USER 1000:1000 +ENTRYPOINT ["python", "-I", "/app/transcribe.py"] diff --git a/ApiCore-Whisper/fetch_model.py b/ApiCore-Whisper/fetch_model.py new file mode 100644 index 0000000..b46fc0b --- /dev/null +++ b/ApiCore-Whisper/fetch_model.py @@ -0,0 +1,69 @@ +"""Downloads the Whisper model files at build time and verifies each one. + +Used only in the Dockerfile (``model`` stage). The revision is a fixed commit +of the Hugging Face repository and every file has a pinned SHA-256: a changed +or tampered file stops the build. At runtime the container has no network and +``HF_HUB_OFFLINE=1``. + +Model: Systran/faster-whisper-small (CTranslate2 conversion of OpenAI Whisper +``small``, MIT licence). The weights are stored as float16; faster-whisper +quantises them to int8 on load (``compute_type='int8'``). +""" + +from __future__ import annotations + +import hashlib +import sys +import urllib.request +from pathlib import Path + +REPOSITORY = "Systran/faster-whisper-small" +REVISION = "536b0662742c02347bc0e980a01041f333bce120" +BASE_URL = f"https://huggingface.co/{REPOSITORY}/resolve/{REVISION}" +FILES: dict[str, str] = { + "config.json": "b55496ac7940a7ae47d2c01eab40edfd8701feec1229d9cce3b40014383fb828", + "model.bin": "3e305921506d8872816023e4c273e75d2419fb89b24da97b4fe7bce14170d671", + "tokenizer.json": "fb7b63191e9bb045082c79fd742a3106a12c99513ab30df4a0d47fa6cb6fd0ab", + "vocabulary.txt": "34ce3fe1c5041027b3f8d42912270993f986dbc4bb34cf27f951e34a1e453913", +} +CHUNK_BYTES = 1024 * 1024 +TIMEOUT_SEC = 120 + + +class ChecksumMismatch(Exception): + """A downloaded file does not have the pinned SHA-256.""" + + +def sha256_of(path: Path) -> str: + digest = hashlib.sha256() + with path.open("rb") as handle: + for chunk in iter(lambda: handle.read(CHUNK_BYTES), b""): + digest.update(chunk) + return digest.hexdigest() + + +def verify(path: Path, expected: str) -> None: + actual = sha256_of(path) + if actual != expected: + raise ChecksumMismatch(f"{path.name}: expected {expected}, got {actual}") + + +def download(url: str, target: Path) -> None: + if not url.startswith("https://"): + raise ValueError("only https downloads are allowed") + with urllib.request.urlopen(url, timeout=TIMEOUT_SEC) as response, target.open("wb") as out: # noqa: S310 + for chunk in iter(lambda: response.read(CHUNK_BYTES), b""): + out.write(chunk) + + +def fetch_all(target_dir: Path, files: dict[str, str] = FILES, base_url: str = BASE_URL) -> None: + target_dir.mkdir(parents=True, exist_ok=True) + for name, expected in files.items(): + target = target_dir / name + download(f"{base_url}/{name}", target) + verify(target, expected) + print(f"ok {name}") + + +if __name__ == "__main__": + fetch_all(Path(sys.argv[1] if len(sys.argv) > 1 else "/models/small")) diff --git a/ApiCore-Whisper/requirements-test.txt b/ApiCore-Whisper/requirements-test.txt new file mode 100644 index 0000000..631a9e4 --- /dev/null +++ b/ApiCore-Whisper/requirements-test.txt @@ -0,0 +1,10 @@ +# Nur fuer die Test-Stufe des Dockerfiles (nicht im Laufzeit-Image). +# Zusammen mit requirements.txt installieren; alle Pakete >= 14 Tage alt (Stichtag 2026-10-02). +pytest==9.1.1 \ + --hash=sha256:37a86b45efb9a47a61a36449063e8e18d0cab3161329fc099eb21783169c4f0c +pluggy==1.6.0 \ + --hash=sha256:e920276dd6813095e9377c0bc5566d94c932c33b27a3e3945d8389c374dd4746 +iniconfig==2.3.0 \ + --hash=sha256:f631c04d2c48c52b84d0d0549c99ff3859c98df65b3101406327ecc7d53fbf12 +pygments==2.21.0 \ + --hash=sha256:2363c69b61c4a97c838da3b130dcd6468f4848992b21a82f2a63ec34377137d9 diff --git a/ApiCore-Whisper/requirements.txt b/ApiCore-Whisper/requirements.txt new file mode 100644 index 0000000..4661b89 --- /dev/null +++ b/ApiCore-Whisper/requirements.txt @@ -0,0 +1,62 @@ +# Gepinnt mit Hashes (pip install --require-hashes --only-binary=:all:). +# Zielplattform: linux/amd64, CPython 3.12 (manylinux-Wheels). Stichtag +# 2026-10-16: jedes Paket war da seit mindestens 14 Tage veroeffentlicht +# (Veroeffentlichung vor 2026-10-02). Kein sdist, nichts wird gebaut. +# +# av ist bewusst 18.1.0 und nicht 19.x: faster-whisper 1.2.1 ruft +# av.open(..., metadata_errors="ignore") auf, das PyAV 19 nicht mehr kennt +# (TypeError beim Dekodieren). Vor dem Anheben von faster-whisper oder av +# mit einer echten Abschrift pruefen. +# +# Neu erzeugen: pip install --dry-run --report r.json --uploaded-prior-to +# --platform manylinux_2_28_x86_64 --python-version 3.12 +# --only-binary=:all: faster-whisper== ergibt die Versionen; die Hashes sind +# die sha256-Werte der passenden Wheels auf PyPI. Danach `pip download +# --no-deps --require-hashes` prueft Hashes und Verfuegbarkeit. +anyio==4.15.1 \ + --hash=sha256:6152fdbbf9a77fdec97731721bebf7c4c44f7c29b424b0065826173efc7ed101 +av==18.1.0 \ + --hash=sha256:8a032e8d8ebc73dec079364b9b4a6837638a2d106e8472314e685ffbf163e700 +certifi==2026.7.22 \ + --hash=sha256:62f22742b58a1a33014a2b6b706588a8d7e2a88ae7bd1a6ebe8c992928483775 +click==8.5.0 \ + --hash=sha256:255bc9599cf7748b4b1a446ccc735421bd08a2ae529a8b88597d3de5664ee360 +ctranslate2==4.8.2 \ + --hash=sha256:554a0b2abf098a11c66ef7930c0a05b3de683a455e8ea3f2e6ab6966aaabc136 +faster-whisper==1.2.1 \ + --hash=sha256:79a66ad50688c0b794dd501dc340a736992a6342f7f95e5811be60b5224a26a7 +filelock==4.0.9 \ + --hash=sha256:9287fd61b99a806e5202be29a83034c0808a1b9830e820537c2f5773981f7eeb +flatbuffers==25.12.19 \ + --hash=sha256:7634f50c427838bb021c2d66a3d1168e9d199b0607e6329399f04846d42e20b4 +fsspec==2026.9.0 \ + --hash=sha256:8dd6e646e99ea382bd85f97a45e6b526a442d79423a7dc673f1e2756d05fcb5f +h11==0.16.0 \ + --hash=sha256:63cf8bbe7522de3bf65932fda1d9c2772064ffb3dae62d55932da54b31cb6c86 +hf-xet==1.6.0 \ + --hash=sha256:d62671bb130879cef0ee4c9ebe47a14af6c66ec53e6d84dc15936e5ffdfac82f +httpcore==1.0.9 \ + --hash=sha256:2d400746a40668fc9dec9810239072b40b4484b640a8c38fd654a024c7a1bf55 +httpx==0.28.1 \ + --hash=sha256:d909fcccc110f8c7faf814ca82a9a4d816bc5a6dbfea25d6591d6985b8ba59ad +huggingface-hub==1.33.0 \ + --hash=sha256:04e434b06e100eddbce9a6e817d72693a7884b10a79bd67ab48080d5c07eb899 +idna==3.20 \ + --hash=sha256:ab7ae7122974553370f0bdb919e1a960b2cd1bc1ef0276416d896db81c14582c +numpy==2.5.3 \ + --hash=sha256:b7e18c623bb5c95acb3b3328861272816ba199fb531921c5d6d0b675f1fde9e3 +onnxruntime==1.30.0 \ + --hash=sha256:fa688e7891a6aa206636fe7372e27ee75fd17713289f6b4fc7b190e0a7de9328 +packaging==26.3 \ + --hash=sha256:d7193f7c8e4e93f444fde0262bf90af30e16fa0ad0ad44cb553c87339b23cd1c +protobuf==7.36.2 \ + --hash=sha256:89f23aa53c24553a2416fd4fd1ec06f74fa42b14b546d8883128813f775bbfd2 \ + --hash=sha256:bdb3a345d48db958e6ce1f18e508beb0cc981d64f24088427549c866cd039f1e +pyyaml==6.0.3 \ + --hash=sha256:ba1cc08a7ccde2d2ec775841541641e4548226580ab850948cbfda66a1befcdc +tokenizers==0.23.2 \ + --hash=sha256:41c2f84d172449b4dadb9cdc508e3e364076613c35b16e76ecfe47a60d1e3305 +tqdm==4.70.1 \ + --hash=sha256:c293e525e6fef9c20e8728fd4612df02a0aa31bb5fe91ecd93e123b1b7bffa73 +typing-extensions==4.16.0 \ + --hash=sha256:481caa481374e813c1b176ada14e97f1f67a4539ce9cfeb3f350d78d6370c2e8 diff --git a/ApiCore-Whisper/tests/test_fetch_model.py b/ApiCore-Whisper/tests/test_fetch_model.py new file mode 100644 index 0000000..abbcbe8 --- /dev/null +++ b/ApiCore-Whisper/tests/test_fetch_model.py @@ -0,0 +1,47 @@ +"""Tests for the build-time model download (no network).""" + +from __future__ import annotations + +import hashlib +import sys +from pathlib import Path + +import pytest + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +import fetch_model # noqa: E402 + + +def test_verify_accepts_the_pinned_hash(tmp_path: Path): + file = tmp_path / "a.bin" + file.write_bytes(b"model") + fetch_model.verify(file, hashlib.sha256(b"model").hexdigest()) + + +def test_verify_rejects_a_changed_file(tmp_path: Path): + file = tmp_path / "a.bin" + file.write_bytes(b"tampered") + with pytest.raises(fetch_model.ChecksumMismatch): + fetch_model.verify(file, hashlib.sha256(b"model").hexdigest()) + + +def test_download_refuses_plain_http(tmp_path: Path): + with pytest.raises(ValueError): + fetch_model.download("http://example.invalid/x", tmp_path / "x") + + +def test_every_pinned_file_has_a_sha256_and_the_revision_is_a_commit(): + assert len(fetch_model.REVISION) == 40 + assert set(fetch_model.FILES) >= {"model.bin", "config.json", "tokenizer.json", "vocabulary.txt"} + for digest in fetch_model.FILES.values(): + assert len(digest) == 64 and int(digest, 16) >= 0 + + +def test_fetch_all_stops_on_a_mismatch(tmp_path: Path, monkeypatch): + def fake_download(url: str, target: Path) -> None: + target.write_bytes(b"evil") + + monkeypatch.setattr(fetch_model, "download", fake_download) + with pytest.raises(fetch_model.ChecksumMismatch): + fetch_model.fetch_all(tmp_path / "m", {"model.bin": hashlib.sha256(b"good").hexdigest()}) diff --git a/ApiCore-Whisper/tests/test_transcribe.py b/ApiCore-Whisper/tests/test_transcribe.py new file mode 100644 index 0000000..b70b899 --- /dev/null +++ b/ApiCore-Whisper/tests/test_transcribe.py @@ -0,0 +1,365 @@ +"""Tests for transcribe.py with a fake model (no model files, no network).""" + +from __future__ import annotations + +import json +import os +import socket +import stat +import sys +import wave +from pathlib import Path +from types import SimpleNamespace + +import pytest + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +import transcribe # noqa: E402 + +SAMPLE_RATE = 8000 + + +FORMATS = {"webm": ("webm", "libopus"), "ogg": ("ogg", "libopus"), "mp4": ("mp4", "aac")} +RATE = 48000 +FRAME = 960 + + +def write_wav(path: Path, seconds: float) -> None: + """Writes silence as real audio in the container the extension names.""" + import av + import numpy as np + + container_name, codec = FORMATS[path.suffix.lstrip(".")] + out = av.open(str(path), "w", format=container_name) + stream = out.add_stream(codec, rate=RATE) + stream.layout = "mono" + frame = av.AudioFrame.from_ndarray(np.zeros((1, FRAME), dtype=np.float32), format="flt", layout="mono") + frame.sample_rate = RATE + for index in range(int(seconds * RATE / FRAME)): + frame.pts = index * FRAME + for packet in stream.encode(frame): + out.mux(packet) + for packet in stream.encode(None): + out.mux(packet) + out.close() + + +class FakeModel: + """Stands in for WhisperModel. ``texts`` is consumed one entry per call, so + the n-th transcribed file gets the n-th entry (the audio is an ndarray).""" + + def __init__(self, texts=None, language="de", probability=1.0): + self.calls: list[int] = [] + self.kwargs: list[dict] = [] + self.audio: list = [] + self.texts = list(texts or []) + self.language = language + self.probability = probability + + def transcribe(self, audio, **kwargs): + self.calls.append(len(self.audio)) + self.audio.append(audio) + self.kwargs.append(kwargs) + entry = self.texts[len(self.calls) - 1] if len(self.calls) <= len(self.texts) else [] + segments = [ + SimpleNamespace(start=index * 2.0, end=index * 2.0 + 1.5, text=f" {text} ") + for index, text in enumerate(entry) + ] + info = SimpleNamespace(language=self.language, language_probability=self.probability) + return iter(segments), info + + +@pytest.fixture +def work(tmp_path: Path) -> Path: + (tmp_path / "in").mkdir(mode=0o700) + (tmp_path / "out").mkdir(mode=0o700) + return tmp_path + + +def make_manifest(work: Path, names: list[str]) -> list[dict]: + items = [] + for name in names: + capture_id = name.split(".")[0] + items.append({"captureId": capture_id, "file": name}) + manifest = { + "v": 1, + "items": [ + {"captureId": i["captureId"], "claimToken": "tok", "file": i["file"], + "mime": "audio/webm", "reportedSec": None} + for i in items + ], + } + (work / "in" / "manifest.json").write_text(json.dumps(manifest), encoding="utf-8") + return items + + +def read_result(work: Path, capture_id: str) -> dict: + return json.loads((work / "out" / f"{capture_id}.json").read_text(encoding="utf-8")) + + +def run_all(work: Path, model: FakeModel, names: list[str]) -> dict: + items = make_manifest(work, names) + return transcribe.run(work, lambda: model, "Durchsicht: Koenigin.", items) + + +def test_writes_an_empty_result_for_silence(work: Path): + write_wav(work / "in" / "cap1.webm", 3) + counters = run_all(work, FakeModel(), ["cap1.webm"]) + result = read_result(work, "cap1") + assert result["error"] == "NO_SPEECH" + assert result["text"] == "" + assert result["segments"] == [] + assert counters["no_speech"] == 1 + + +def test_rejects_audio_longer_than_240_seconds(work: Path): + write_wav(work / "in" / "long1.webm", 241) + model = FakeModel(texts=[["das darf nie ankommen"]]) + counters = run_all(work, model, ["long1.webm"]) + result = read_result(work, "long1") + assert result["error"] == "TOO_LONG" + assert result["durationSec"] > 240 + assert model.calls == [] # never reaches the model + assert counters["too_long"] == 1 + + +def test_accepts_audio_of_exactly_the_limit_region(work: Path): + write_wav(work / "in" / "ok1.webm", 239) + model = FakeModel(texts=[["Die Koenigin ist da"]]) + run_all(work, model, ["ok1.webm"]) + assert "error" not in read_result(work, "ok1") + + +def test_rejects_a_corrupt_file_without_crashing_the_batch(work: Path): + (work / "in" / "bad1.webm").write_bytes(os.urandom(4096)) + write_wav(work / "in" / "good1.webm", 5) + model = FakeModel(texts=[["Zwei Brutwaben mit Stiften"]]) + counters = run_all(work, model, ["bad1.webm", "good1.webm"]) + assert read_result(work, "bad1")["error"] == "DECODE_FAILED" + assert read_result(work, "good1")["text"] == "Zwei Brutwaben mit Stiften" + assert len(model.calls) == 1 + assert counters["failed"] == 1 and counters["transcribed"] == 1 + + +def test_a_model_error_on_one_file_does_not_stop_the_others(work: Path): + class Exploding(FakeModel): + def transcribe(self, audio, **kwargs): + if not self.calls and not self.audio: + self.audio.append(None) + self.calls.append(0) + raise RuntimeError("secret transcript text must not leak") + return super().transcribe(audio, **kwargs) + + write_wav(work / "in" / "a1.webm", 5) + write_wav(work / "in" / "b1.webm", 5) + model = Exploding(texts=[[], ["Volk ist ruhig und sanftmuetig"]]) + run_all(work, model, ["a1.webm", "b1.webm"]) + assert read_result(work, "a1")["error"] == "DECODE_FAILED" + assert read_result(work, "b1")["text"].startswith("Volk") + + +def test_processes_files_one_after_the_other_and_writes_one_result_each(work: Path): + names = ["n1.webm", "n2.webm", "n3.webm"] + for name in names: + write_wav(work / "in" / name, 4) + model = FakeModel(texts=[[f"Notiz fuer {name}"] for name in names]) + counters = run_all(work, model, names) + assert len(model.calls) == 3 + for name in names: + assert read_result(work, name.split(".")[0])["text"] == f"Notiz fuer {name}" + assert sorted(p.name for p in (work / "out").iterdir()) == ["n1.json", "n2.json", "n3.json"] + assert counters["transcribed"] == 3 + + +def test_loads_the_model_once_for_the_whole_batch(work: Path): + names = ["m1.webm", "m2.webm"] + for name in names: + write_wav(work / "in" / name, 4) + loads = [] + model = FakeModel(texts=[["Genug Text hier"] for _ in names]) + + def factory(): + loads.append(1) + return model + + transcribe.run(work, factory, "p", make_manifest(work, names)) + assert len(loads) == 1 + + +def test_does_not_read_the_network(monkeypatch): + monkeypatch.setattr(socket, "socket", socket.socket) # restored after the test + transcribe.block_network() + with pytest.raises(OSError): + socket.socket(socket.AF_INET, socket.SOCK_STREAM) + with pytest.raises(OSError): + socket.create_connection(("127.0.0.1", 9), timeout=1) + + +def test_the_network_block_is_a_subclassable_class(monkeypatch): + monkeypatch.setattr(socket, "socket", socket.socket) + transcribe.block_network() + + class Later(socket.socket): # what ssl does at import time + pass + + with pytest.raises(OSError): + Later() + + +def test_writes_segments_with_start_end_and_text(work: Path): + write_wav(work / "in" / "seg1.webm", 10) + model = FakeModel(texts=[["Erster Satz", "Zweiter Satz"]]) + run_all(work, model, ["seg1.webm"]) + result = read_result(work, "seg1") + assert result["v"] == 1 + assert result["captureId"] == "seg1" + assert result["text"] == "Erster Satz Zweiter Satz" + assert result["segments"] == [ + {"start": 0.0, "end": 1.5, "text": "Erster Satz"}, + {"start": 2.0, "end": 3.5, "text": "Zweiter Satz"}, + ] + assert result["language"] == "de" + assert result["languageProbability"] == 1.0 + assert 9.9 <= result["durationSec"] <= 10.1 + + +def test_passes_the_agreed_whisper_settings(work: Path): + write_wav(work / "in" / "cfg1.webm", 4) + model = FakeModel(texts=[["Genug Text hier"]]) + run_all(work, model, ["cfg1.webm"]) + assert model.kwargs[0] == { + "language": "de", + "beam_size": 1, + "vad_filter": True, + "condition_on_previous_text": False, + "initial_prompt": "Durchsicht: Koenigin.", + } + + +def test_a_text_under_five_characters_is_no_speech(work: Path): + write_wav(work / "in" / "short1.webm", 4) + run_all(work, FakeModel(texts=[["Ja"]]), ["short1.webm"]) + assert read_result(work, "short1")["error"] == "NO_SPEECH" + + +def test_another_language_with_low_probability_is_no_speech(work: Path): + write_wav(work / "in" / "fr1.webm", 4) + model = FakeModel(texts=[["Bonjour tout le monde"]], language="fr", probability=0.4) + run_all(work, model, ["fr1.webm"]) + assert read_result(work, "fr1")["error"] == "NO_SPEECH" + + +def test_another_language_with_high_probability_keeps_the_text(work: Path): + write_wav(work / "in" / "en1.webm", 4) + model = FakeModel(texts=[["Hello there everybody"]], language="en", probability=0.9) + run_all(work, model, ["en1.webm"]) + assert "error" not in read_result(work, "en1") + + +def test_no_manifest_means_exit_0_without_loading_the_model(tmp_path: Path, monkeypatch): + def forbidden(*args, **kwargs): + raise AssertionError("the model must not load without a manifest") + + monkeypatch.setattr(transcribe, "load_whisper_model", forbidden) + assert transcribe.main({"VOICE_WORK_DIR": str(tmp_path)}) == 0 + + +def test_an_empty_manifest_does_not_load_the_model(work: Path, monkeypatch): + (work / "in" / "manifest.json").write_text('{"v":1,"items":[]}', encoding="utf-8") + monkeypatch.setattr(transcribe, "load_whisper_model", lambda *a: pytest.fail("loaded")) + assert transcribe.main({"VOICE_WORK_DIR": str(work)}) == 0 + + +def test_a_broken_manifest_fails_the_run(work: Path): + (work / "in" / "manifest.json").write_text("not json", encoding="utf-8") + assert transcribe.main({"VOICE_WORK_DIR": str(work)}) == 1 + + +@pytest.mark.parametrize( + "item", + [ + {"captureId": "../x", "file": "../x.webm"}, + {"captureId": "a1", "file": "../a1.webm"}, + {"captureId": "a1", "file": "other.webm"}, + {"captureId": "a1", "file": "a1.wav"}, + {"captureId": "", "file": ".webm"}, + ], +) +def test_a_manifest_item_cannot_leave_the_work_dir(work: Path, item: dict): + manifest = {"v": 1, "items": [item]} + (work / "in" / "manifest.json").write_text(json.dumps(manifest), encoding="utf-8") + with pytest.raises(transcribe.ManifestInvalid): + transcribe.load_manifest(work) + + +def test_a_missing_input_file_is_decode_failed(work: Path): + run_all(work, FakeModel(), ["gone1.webm"]) + assert read_result(work, "gone1")["error"] == "DECODE_FAILED" + + +@pytest.mark.skipif(os.name == "nt", reason="POSIX file modes") +def test_result_files_are_private(work: Path): + write_wav(work / "in" / "priv1.webm", 4) + run_all(work, FakeModel(texts=[["Genug Text hier"]]), ["priv1.webm"]) + mode = stat.S_IMODE((work / "out" / "priv1.json").stat().st_mode) + assert mode == 0o600 + + +def test_output_never_prints_the_transcript(work: Path, capsys, monkeypatch): + write_wav(work / "in" / "quiet1.webm", 4) + make_manifest(work, ["quiet1.webm"]) + model = FakeModel(texts=[["Geheimer Inhalt der Notiz"]]) + vocabulary = work / "vocabulary.txt" + vocabulary.write_text("Durchsicht: Koenigin.\n", encoding="utf-8") + monkeypatch.setattr(transcribe, "load_whisper_model", lambda *a: model) + code = transcribe.main({"VOICE_WORK_DIR": str(work), "WHISPER_VOCABULARY": str(vocabulary)}) + captured = capsys.readouterr() + assert code == 0 + assert "Geheimer" not in captured.out + captured.err + assert captured.out.strip() == "transcribed=1 no_speech=0 too_long=0 failed=0" + + +def test_cpu_threads_are_validated(): + assert transcribe.cpu_threads({}) == 2 + assert transcribe.cpu_threads({"WHISPER_CPU_THREADS": "4"}) == 4 + for bad in ("0", "abc", "99", "-1"): + with pytest.raises(ValueError): + transcribe.cpu_threads({"WHISPER_CPU_THREADS": bad}) + + +@pytest.mark.parametrize("extension", ["webm", "ogg", "mp4"]) +def test_decodes_each_container_and_hands_the_model_16khz_float_samples(work: Path, extension: str): + write_wav(work / "in" / f"fmt1.{extension}", 3) + model = FakeModel(texts=[["Genug Text hier"]]) + run_all(work, model, [f"fmt1.{extension}"]) + assert "error" not in read_result(work, "fmt1") + audio = model.audio[0] + assert audio.dtype.name == "float32" and audio.ndim == 1 + assert 2.8 <= audio.size / 16000 <= 3.3 + + +def test_a_file_in_another_container_than_its_extension_is_decode_failed(work: Path): + write_wav(work / "in" / "ogg1.ogg", 2) + (work / "in" / "mixed1.webm").write_bytes((work / "in" / "ogg1.ogg").read_bytes()) + model = FakeModel(texts=[["x" * 10]]) + run_all(work, model, ["mixed1.webm"]) + assert read_result(work, "mixed1")["error"] == "DECODE_FAILED" + assert model.calls == [] + + +def test_audio_without_any_measurable_length_is_decode_failed(work: Path): + write_wav(work / "in" / "empty1.webm", 0) + model = FakeModel(texts=[["x" * 10]]) + run_all(work, model, ["empty1.webm"]) + assert read_result(work, "empty1")["error"] == "DECODE_FAILED" + assert model.calls == [] + + +def test_the_decoded_length_beats_the_demuxed_length(work: Path, monkeypatch): + write_wav(work / "in" / "liar1.webm", 245) + monkeypatch.setattr(transcribe, "measure_duration", lambda *a, **k: 10.0) + model = FakeModel(texts=[["x" * 10]]) + run_all(work, model, ["liar1.webm"]) + assert read_result(work, "liar1")["error"] == "TOO_LONG" + assert model.calls == [] diff --git a/ApiCore-Whisper/transcribe.py b/ApiCore-Whisper/transcribe.py new file mode 100644 index 0000000..9b5c6e3 --- /dev/null +++ b/ApiCore-Whisper/transcribe.py @@ -0,0 +1,363 @@ +"""Batch transcription for the voice CronJob (Task 13, spec 6.3 and 9.2). + +Second init container of ``apicore-job-voice-transcribe``. It reads +``/in/manifest.json`` written by the ``fetch`` phase, transcribes the +listed audio files one after the other with one loaded model and writes one +``/out/.json`` per file for the ``ingest`` phase. + +The container has no secrets, no database and no network. Belt and braces to +the NetworkPolicy: ``socket.socket`` is blocked before the model loads. +Transcripts and audio are never printed or logged; stdout carries counters +only, stderr only exception class names. + +Result file (matches ``WhisperResult`` in the backend ``work-dir.ts``):: + + {v: 1, captureId, text, durationSec, language, languageProbability, + segments: [{start, end, text}], error?: NO_SPEECH | TOO_LONG | DECODE_FAILED} +""" + +from __future__ import annotations + +import json +import os +import re +import socket +import stat +import sys +from collections.abc import Callable, Iterable +from pathlib import Path +from typing import Any + +RESULT_VERSION = 1 +MAX_AUDIO_SEC = 240.0 +MIN_TEXT_CHARS = 5 +GERMAN = "de" +MIN_FOREIGN_LANGUAGE_PROBABILITY = 0.6 +MAX_AUDIO_BYTES = 3 * 1024 * 1024 +MAX_MANIFEST_BYTES = 256 * 1024 +MAX_ITEMS = 50 +DEFAULT_WORK_DIR = "/work" +DEFAULT_MODEL_DIR = "/models/small" +DEFAULT_CPU_THREADS = 2 +MAX_CPU_THREADS = 16 +UNKNOWN_LANGUAGE = "und" +SAMPLE_RATE = 16000 +# Container per file extension (the manifest only allows these three). Naming +# the demuxer keeps libav from probing every demuxer on untrusted bytes. +CONTAINER_FORMATS = {"webm": "matroska", "ogg": "ogg", "mp4": "mov"} + +ID_PATTERN = re.compile(r"^[A-Za-z0-9_-]{1,64}$") +FILE_PATTERN = re.compile(r"^([A-Za-z0-9_-]{1,64})\.(webm|ogg|mp4)$") + + +class DecodeFailed(Exception): + """The file is not usable audio. Carries no content.""" + + +class ManifestInvalid(Exception): + """The manifest exists but is not valid.""" + + +class _BlockedSocket(socket.socket): + """A socket that cannot be created (a class, so later subclassing works).""" + + def __init__(self, *args: Any, **kwargs: Any) -> None: # noqa: D107 + raise OSError("network access is disabled in the whisper container") + + +def block_network() -> None: + """Make every later ``socket.socket(...)`` fail.""" + socket.socket = _BlockedSocket # type: ignore[misc,assignment] + + +# --- audio --------------------------------------------------------------- + + +def container_format(path: Path) -> str: + try: + return CONTAINER_FORMATS[path.suffix.lstrip(".")] + except KeyError: + raise DecodeFailed("unsupported container") from None + + +def decode_audio(path: Path, limit: float = MAX_AUDIO_SEC) -> Any: + """Mono float32 samples at 16 kHz (what Whisper takes), decoded here so the + demuxer is named explicitly. Stops a little past ``limit`` seconds.""" + import av + import numpy as np + + max_samples = int((limit + 1) * SAMPLE_RATE) + try: + with av.open(str(path), mode="r", format=container_format(path)) as container: + stream = next((s for s in container.streams if s.type == "audio"), None) + if stream is None: + raise DecodeFailed("no audio stream") + resampler = av.AudioResampler(format="s16", layout="mono", rate=SAMPLE_RATE) + chunks = [] + total = 0 + for frame in container.decode(stream): + for out in resampler.resample(frame): + chunk = out.to_ndarray().reshape(-1) + chunks.append(chunk) + total += chunk.size + if total > max_samples: + break + for out in resampler.resample(None): + chunks.append(out.to_ndarray().reshape(-1)) + except DecodeFailed: + raise + except Exception as error: + raise DecodeFailed(type(error).__name__) from None + if not chunks: + raise DecodeFailed("no samples") + return np.concatenate(chunks).astype(np.float32) / 32768.0 + + +def measure_duration(path: Path, limit: float = MAX_AUDIO_SEC) -> float: + """Measured length in seconds, by demuxing packets (no decoding). + + Stops as soon as the length passes ``limit`` (the value is then a lower + bound above the limit). The container header is not trusted: browser + recordings often have none, and a forged one must not hide a long file. + """ + import av # imported here so an idle run (no manifest) stays fast + + try: + with av.open(str(path), mode="r", format=container_format(path)) as container: + stream = next((s for s in container.streams if s.type == "audio"), None) + if stream is None: + raise DecodeFailed("no audio stream") + end = 0.0 + counted = 0.0 + for packet in container.demux(stream): + if packet.size == 0 or packet.pts is None or packet.time_base is None: + if packet.duration is not None and packet.time_base is not None: + counted += float(packet.duration * packet.time_base) + end = max(end, counted) + continue + packet_end = float((packet.pts + (packet.duration or 0)) * packet.time_base) + end = max(end, packet_end) + counted = end + if end > limit: + break + if end <= 0.0: + raise DecodeFailed("no measurable duration") + return round(end, 2) + except DecodeFailed: + raise + except Exception as error: # av.error.FFmpegError, OSError, ValueError + raise DecodeFailed(type(error).__name__) from None + + +# --- results ------------------------------------------------------------- + + +def error_result(capture_id: str, error: str, duration: float = 0.0) -> dict[str, Any]: + return { + "v": RESULT_VERSION, + "captureId": capture_id, + "text": "", + "durationSec": duration, + "language": UNKNOWN_LANGUAGE, + "languageProbability": 0.0, + "segments": [], + "error": error, + } + + +def _clamp_probability(value: Any) -> float: + try: + number = float(value) + except (TypeError, ValueError): + return 0.0 + if number != number: # NaN + return 0.0 + return round(min(max(number, 0.0), 1.0), 3) + + +def build_result( + capture_id: str, duration: float, segments: Iterable[Any], info: Any +) -> dict[str, Any]: + parts = [] + for segment in segments: + text = str(segment.text).strip() + if text: + parts.append( + { + "start": round(float(segment.start), 2), + "end": round(float(segment.end), 2), + "text": text, + } + ) + full_text = " ".join(part["text"] for part in parts) + language = str(getattr(info, "language", "") or "") + if not re.fullmatch(r"[A-Za-z-]{1,16}", language): + language = UNKNOWN_LANGUAGE + probability = _clamp_probability(getattr(info, "language_probability", 0.0)) + foreign = language != GERMAN and probability < MIN_FOREIGN_LANGUAGE_PROBABILITY + if len(full_text) < MIN_TEXT_CHARS or foreign: + result = error_result(capture_id, "NO_SPEECH", duration) + result["language"] = language + result["languageProbability"] = probability + return result + return { + "v": RESULT_VERSION, + "captureId": capture_id, + "text": full_text, + "durationSec": duration, + "language": language, + "languageProbability": probability, + "segments": parts, + } + + +def write_result(out_dir: Path, capture_id: str, result: dict[str, Any]) -> None: + """Writes by rename, mode 0600, so ingest never sees a half-written file.""" + target = out_dir / f"{capture_id}.json" + temporary = out_dir / f"{capture_id}.json.tmp" + flags = os.O_WRONLY | os.O_CREAT | os.O_TRUNC + descriptor = os.open(temporary, flags, 0o600) + with os.fdopen(descriptor, "w", encoding="utf-8") as handle: + json.dump(result, handle, ensure_ascii=False) + os.replace(temporary, target) + + +# --- manifest -------------------------------------------------------------- + + +def load_manifest(work_dir: Path) -> list[dict[str, str]] | None: + """Items of this run; None if there is no manifest (nothing to do).""" + path = work_dir / "in" / "manifest.json" + try: + info = path.lstat() + except FileNotFoundError: + return None + if not stat.S_ISREG(info.st_mode) or info.st_size > MAX_MANIFEST_BYTES: + raise ManifestInvalid("manifest is not a regular file or too large") + try: + parsed = json.loads(path.read_text(encoding="utf-8")) + except (OSError, ValueError): + raise ManifestInvalid("manifest is not readable JSON") from None + items = parsed.get("items") if isinstance(parsed, dict) else None + if not isinstance(items, list) or parsed.get("v") != 1 or len(items) > MAX_ITEMS: + raise ManifestInvalid("manifest has an unexpected shape") + checked: list[dict[str, str]] = [] + for item in items: + capture_id = item.get("captureId") if isinstance(item, dict) else None + file_name = item.get("file") if isinstance(item, dict) else None + match = FILE_PATTERN.match(file_name) if isinstance(file_name, str) else None + if not isinstance(capture_id, str) or not ID_PATTERN.match(capture_id): + raise ManifestInvalid("manifest item without a usable id") + if match is None or match.group(1) != capture_id: + raise ManifestInvalid("manifest item with a file name that is not its id") + checked.append({"captureId": capture_id, "file": file_name}) + return checked + + +# --- run ------------------------------------------------------------------- + +Counters = dict[str, int] +ModelFactory = Callable[[], Any] + + +def _usable_input(path: Path) -> bool: + try: + info = path.lstat() + except OSError: + return False + return stat.S_ISREG(info.st_mode) and 0 < info.st_size <= MAX_AUDIO_BYTES + + +def transcribe_one(model: Any, path: Path, capture_id: str, prompt: str) -> dict[str, Any]: + """One file to one result. Never raises: a broken file is DECODE_FAILED.""" + if not _usable_input(path): + return error_result(capture_id, "DECODE_FAILED") + try: + duration = measure_duration(path) + except DecodeFailed: + return error_result(capture_id, "DECODE_FAILED") + if duration > MAX_AUDIO_SEC: + return error_result(capture_id, "TOO_LONG", duration) + try: + audio = decode_audio(path) + if audio.size / SAMPLE_RATE > MAX_AUDIO_SEC: # decoded length beats any header + return error_result(capture_id, "TOO_LONG", round(audio.size / SAMPLE_RATE, 2)) + segments, info = model.transcribe( + audio, + language=GERMAN, + beam_size=1, + vad_filter=True, + condition_on_previous_text=False, + initial_prompt=prompt, + ) + collected = list(segments) # the generator does the work here + return build_result(capture_id, duration, collected, info) + except DecodeFailed: + return error_result(capture_id, "DECODE_FAILED", duration) + except Exception as error: # decoder or model error on this file only + print(f"transcribe failed: {type(error).__name__}", file=sys.stderr) + return error_result(capture_id, "DECODE_FAILED", duration) + + +def run( + work_dir: Path, + model_factory: ModelFactory, + prompt: str, + items: list[dict[str, str]], +) -> Counters: + counters: Counters = {"transcribed": 0, "no_speech": 0, "too_long": 0, "failed": 0} + out_dir = work_dir / "out" + out_dir.mkdir(mode=0o700, exist_ok=True) + model = model_factory() + for item in items: # one after the other: the pod has one CPU budget + result = transcribe_one(model, work_dir / "in" / item["file"], item["captureId"], prompt) + write_result(out_dir, item["captureId"], result) + error = result.get("error") + key = { + None: "transcribed", + "NO_SPEECH": "no_speech", + "TOO_LONG": "too_long", + }.get(error, "failed") + counters[key] += 1 + return counters + + +def cpu_threads(env: dict[str, str]) -> int: + raw = env.get("WHISPER_CPU_THREADS", "") + if not raw: + return DEFAULT_CPU_THREADS + if not raw.isdigit() or not 1 <= int(raw) <= MAX_CPU_THREADS: + raise ValueError(f"WHISPER_CPU_THREADS must be a whole number from 1 to {MAX_CPU_THREADS}") + return int(raw) + + +def load_whisper_model(model_dir: str, threads: int) -> Any: + """Imports faster-whisper, then cuts the network, then loads the model.""" + from faster_whisper import WhisperModel + + block_network() + return WhisperModel(model_dir, device="cpu", compute_type="int8", cpu_threads=threads) + + +def main(env: dict[str, str] | None = None) -> int: + env = dict(os.environ) if env is None else env + work_dir = Path(env.get("VOICE_WORK_DIR") or DEFAULT_WORK_DIR) + try: + items = load_manifest(work_dir) + except ManifestInvalid as error: + print(f"whisper: {error}", file=sys.stderr) + return 1 + if not items: + print("transcribed=0 no_speech=0 too_long=0 failed=0") + return 0 + prompt_file = Path(env.get("WHISPER_VOCABULARY") or Path(__file__).with_name("vocabulary.txt")) + prompt = prompt_file.read_text(encoding="utf-8").strip() + model_dir = env.get("WHISPER_MODEL_DIR") or DEFAULT_MODEL_DIR + threads = cpu_threads(env) + counters = run(work_dir, lambda: load_whisper_model(model_dir, threads), prompt, items) + print(" ".join(f"{name}={count}" for name, count in counters.items())) + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/deploy/k8s/components/voice/cronjob-voice-transcribe.yaml b/deploy/k8s/components/voice/cronjob-voice-transcribe.yaml new file mode 100644 index 0000000..b0bb6e7 --- /dev/null +++ b/deploy/k8s/components/voice/cronjob-voice-transcribe.yaml @@ -0,0 +1,165 @@ +# Abschreiben im Cluster (Spec T7 6.3 und 11.1): alle 2 Minuten ein Pod mit +# drei Containern, die nacheinander laufen: +# fetch (initContainer) claimt Aufnahmen, entschluesselt, legt Ton nach /work/in +# whisper (initContainer) liest /work/in, schreibt /work/out; ohne Netz, +# ohne Schluessel, ohne Datenbank-Zugang +# ingest (Hauptcontainer) liest /work/out, verschluesselt, speichert, loescht Ton +# +# Der gemeinsame CronJob-Patch aus components/backend-roles trifft alle +# CronJobs mit app=apicore-job und wuerde diesem Pod einen Container `job` +# (Backend-Image, alle Secrets) hinzufuegen. Darum traegt das CronJob-Objekt +# app=apicore-voice-job; nur das Pod-Template traegt app=apicore-job (damit die +# NetworkPolicy apicore-jobs gilt: Egress nur Postgres und DNS). Haertung, +# Zeitzone und Forbid stehen deshalb hier vollstaendig. +# +# fetch, whisper und ingest laufen mit derselben uid 1000 und fsGroup 1000: +# /work/in und /work/out sind 0700, die Dateien 0600. +apiVersion: batch/v1 +kind: CronJob +metadata: + name: apicore-job-voice-transcribe + labels: + app: apicore-voice-job + apicore.io/job: voice-transcribe +spec: + schedule: "*/2 * * * *" + timeZone: Europe/Berlin + concurrencyPolicy: Forbid + startingDeadlineSeconds: 120 + successfulJobsHistoryLimit: 3 + failedJobsHistoryLimit: 3 + jobTemplate: + spec: + backoffLimit: 0 + activeDeadlineSeconds: 1500 + ttlSecondsAfterFinished: 86400 + template: + metadata: + labels: + app: apicore-job + apicore.io/job: voice-transcribe + db-client: "true" + spec: + restartPolicy: Never + automountServiceAccountToken: false + enableServiceLinks: false + securityContext: + runAsNonRoot: true + runAsUser: 1000 + runAsGroup: 1000 + fsGroup: 1000 + seccompProfile: + type: RuntimeDefault + volumes: + # Entschluesselter Ton und Abschriften liegen nur im RAM, nie auf + # einer Platte. Das Backend rechnet mit diesen 32Mi (work-dir.ts). + - name: work + emptyDir: + medium: Memory + sizeLimit: 32Mi + # /tmp von fetch und ingest (HOME=/tmp). Whisper bekommt ein eigenes + # Volume: ein gemeinsames /tmp liesse einen kompromittierten + # Decoder Code unter $HOME/.node_modules ablegen, den ingest mit + # allen Secrets laedt. + - name: tmp + emptyDir: + sizeLimit: 64Mi + - name: whisper-tmp + emptyDir: + medium: Memory + sizeLimit: 16Mi + initContainers: + - name: fetch + image: git.encry.de/nick.rosenloecher/apicore-backend + imagePullPolicy: IfNotPresent + command: ["node", "dist/cli.js", "voice-transcribe", "--phase=fetch"] + env: + - name: APICORE_ROLE + value: job + - name: HOME + value: /tmp + envFrom: + - configMapRef: + name: apicore-backend-config + - secretRef: + name: apicore-backend-secret + securityContext: + allowPrivilegeEscalation: false + readOnlyRootFilesystem: true + capabilities: + drop: ["ALL"] + volumeMounts: + - name: work + mountPath: /work + - name: tmp + mountPath: /tmp + resources: + requests: + cpu: 50m + memory: 192Mi + limits: + cpu: 500m + memory: 512Mi + # Kein envFrom, kein secretKeyRef, kein Service-Account-Token. Nur + # die Thread-Zahl kommt von aussen; alles andere steht im Image. + - name: whisper + image: git.encry.de/nick.rosenloecher/apicore-whisper + imagePullPolicy: IfNotPresent + env: + - name: WHISPER_CPU_THREADS + value: "2" + securityContext: + allowPrivilegeEscalation: false + readOnlyRootFilesystem: true + runAsNonRoot: true + runAsUser: 1000 + runAsGroup: 1000 + capabilities: + drop: ["ALL"] + seccompProfile: + type: RuntimeDefault + volumeMounts: + # Der ganze Mount, Schreibrecht fuer /work/out. + - name: work + mountPath: /work + - name: whisper-tmp + mountPath: /tmp + resources: + requests: + cpu: 50m + memory: 1280Mi + limits: + cpu: 2000m + memory: 1280Mi + containers: + - name: ingest + image: git.encry.de/nick.rosenloecher/apicore-backend + imagePullPolicy: IfNotPresent + command: ["node", "dist/cli.js", "voice-transcribe", "--phase=ingest"] + env: + - name: APICORE_ROLE + value: job + - name: HOME + value: /tmp + envFrom: + - configMapRef: + name: apicore-backend-config + - secretRef: + name: apicore-backend-secret + securityContext: + allowPrivilegeEscalation: false + readOnlyRootFilesystem: true + capabilities: + drop: ["ALL"] + volumeMounts: + - name: work + mountPath: /work + - name: tmp + mountPath: /tmp + resources: + requests: + cpu: 50m + memory: 192Mi + limits: + cpu: 500m + memory: 512Mi diff --git a/deploy/k8s/components/voice/kustomization.yaml b/deploy/k8s/components/voice/kustomization.yaml new file mode 100644 index 0000000..c46f75c --- /dev/null +++ b/deploy/k8s/components/voice/kustomization.yaml @@ -0,0 +1,11 @@ +# Sprachnotizen (Spec T7 11.1 und 11.3): der CronJob, der Ton im Cluster +# abschreibt. Rollout: nach backend-roles (gleiche Backend-Rollen, gleiches +# Secret) und erst, wenn VOICE_ENCRYPTION_KEY in apicore-backend-secret steht. +# +# Zurueckgestellt (Controller-Entscheid 9): patch-worker-voice.yaml mit +# ANTHROPIC_API_KEY (secretKeyRef apicore-anthropic, optional) im Worker. Es +# wird erst angelegt, wenn die Claude-Auswertung freigegeben ist. +apiVersion: kustomize.config.k8s.io/v1alpha1 +kind: Component +resources: + - cronjob-voice-transcribe.yaml diff --git a/deploy/k8s/overlays/encry/configmap-backend.yaml b/deploy/k8s/overlays/encry/configmap-backend.yaml index fa9027c..3b082b4 100644 --- a/deploy/k8s/overlays/encry/configmap-backend.yaml +++ b/deploy/k8s/overlays/encry/configmap-backend.yaml @@ -28,3 +28,10 @@ data: # bei BotFather noch nicht angelegt ist (Runbook); dann den Namen eintragen. # Das Token liegt im Secret apicore-telegram (nur Worker). TELEGRAM_BOT_USERNAME: "" + # Sprachnotizen (T7). VOICE_RECORDING_ENABLED blendet die Aufnahme in der App + # aus ("false") oder ein ("true"); VOICE_MODEL und VOICE_DAILY_EXTRACTIONS + # gelten fuer die Claude-Auswertung (zurueckgestellt). Den Schluessel + # ANTHROPIC_API_KEY gibt es hier nie: er gehoert ins Secret apicore-anthropic. + VOICE_RECORDING_ENABLED: "false" + VOICE_MODEL: "claude-sonnet-5-5" + VOICE_DAILY_EXTRACTIONS: "60" diff --git a/deploy/k8s/overlays/encry/kustomization.yaml b/deploy/k8s/overlays/encry/kustomization.yaml index 1f48683..33b0423 100644 --- a/deploy/k8s/overlays/encry/kustomization.yaml +++ b/deploy/k8s/overlays/encry/kustomization.yaml @@ -43,10 +43,15 @@ components: - ../../components/pod-hardening # Schritt 3 - ../../components/backend-roles # Schritt 3 - ../../components/api-hpa # Schritt 5 + - ../../components/voice # Sprachnotizen (T7), erst nach VOICE_ENCRYPTION_KEY im Secret images: - name: git.encry.de/nick.rosenloecher/apicore-backend digest: sha256:0000000000000000000000000000000000000000000000000000000000000000 - name: git.encry.de/nick.rosenloecher/apicore-frontend digest: sha256:0000000000000000000000000000000000000000000000000000000000000000 + # Abschreib-Image (.gitea/workflows/whisper-image.yml, Tag sha-). + # Platzhalter bis zum ersten CI-Build; dann den Digest eintragen. + - name: git.encry.de/nick.rosenloecher/apicore-whisper + digest: sha256:0000000000000000000000000000000000000000000000000000000000000000 - name: postgres digest: sha256:0000000000000000000000000000000000000000000000000000000000000000 diff --git a/deploy/k8s/tests/manifest-policy.test.mjs b/deploy/k8s/tests/manifest-policy.test.mjs index 7e0681a..9960802 100644 --- a/deploy/k8s/tests/manifest-policy.test.mjs +++ b/deploy/k8s/tests/manifest-policy.test.mjs @@ -18,7 +18,7 @@ import { const APP_PODS = ['apicore-backend', 'apicore-worker', 'apicore-job', 'apicore-frontend']; const DB_CLIENT_APPS = ['apicore-backend', 'apicore-worker', 'apicore-job', 'apicore-db-backup']; -const KNOWN_JOBS = ['daily-evaluation', 'task-templates', 'cleanup', 'app-backup', 'rekey']; +const KNOWN_JOBS = ['daily-evaluation', 'task-templates', 'cleanup', 'app-backup', 'rekey', 'voice-transcribe']; const PRIVATE_V4 = ['10.0.0.0/8', '172.16.0.0/12', '192.168.0.0/16', '100.64.0.0/10', '169.254.0.0/16', '127.0.0.0/8']; const PRIVATE_V6 = ['fc00::/7', 'fe80::/10', '::1/128']; const MIGRATE = ['node', 'node_modules/prisma/build/index.js', 'migrate', 'deploy']; @@ -255,7 +255,9 @@ test('all app pods: automountServiceAccountToken false, readOnlyRootFilesystem t test('cronjobs: Forbid, Europe/Berlin, command node dist/cli.js ', () => { const cronJobs = target.docs.filter((d) => d.kind === 'CronJob'); for (const cj of cronJobs) assert.equal(cj.spec.concurrencyPolicy, 'Forbid', `${cj.metadata.name}`); - const appJobs = cronJobs.filter((cj) => podTemplateOf(cj).metadata?.labels?.app === 'apicore-job'); + // The shared job patch (backend-roles) selects CronJob objects labelled app=apicore-job; + // the voice CronJob is deliberately not one of them (see the voice tests below). + const appJobs = cronJobs.filter((cj) => cj.metadata.labels?.app === 'apicore-job'); assert.deepEqual( appJobs.map((cj) => cj.metadata.name).sort(), Object.keys(CRON_JOBS).map((name) => `apicore-job-${name}`).sort(), @@ -541,3 +543,190 @@ for (const [name, model] of [...stages, ['target', target]]) { } }); } + +// --- voice transcription CronJob (spec T7 6.3, 9.2, 11.1; plan task 13) ------ + +const VOICE_CRONJOB = 'apicore-job-voice-transcribe'; +const WHISPER_IMAGE = 'git.encry.de/nick.rosenloecher/apicore-whisper'; +const ANTHROPIC_KEY = 'ANTHROPIC_API_KEY'; +const ANTHROPIC_SECRET = 'apicore-anthropic'; + +function voiceCronJob(model) { + const found = model.docs.filter((d) => d.kind === 'CronJob' && d.metadata.name === VOICE_CRONJOB); + assert.equal(found.length, 1, `expected exactly one CronJob ${VOICE_CRONJOB}`); + return found[0]; +} +const voicePodOf = (model) => { + const pod = model.byOwner(`CronJob/${VOICE_CRONJOB}`); + assert.ok(pod, 'voice pod template missing'); + return pod; +}; +const voiceContainer = (pod, name) => { + const found = [...pod.initContainers, ...pod.containers].filter((c) => c.name === name); + assert.equal(found.length, 1, `expected exactly one container ${name}`); + return found[0]; +}; + +test('whisper container has no env from secrets, no secretKeyRef and no service account token', () => { + const pod = voicePodOf(target); + const whisper = voiceContainer(pod, 'whisper'); + assert.equal(pod.spec.automountServiceAccountToken, false, 'service account token mounted'); + assert.equal(whisper.envFrom, undefined, 'whisper takes envFrom'); + assert.deepEqual(whisper.env, [{ name: 'WHISPER_CPU_THREADS', value: '2' }], 'whisper env is only the thread count'); + assert.ok(!(whisper.env ?? []).some((e) => e.valueFrom), 'whisper env uses valueFrom'); + assert.equal(whisper.image.split('@')[0], WHISPER_IMAGE); + // Every volume of the pod is an emptyDir: no secret, configMap or projected token. + for (const volume of pod.spec.volumes ?? []) { + assert.ok(volume.emptyDir, `volume ${volume.name} is not an emptyDir`); + } + assert.deepEqual( + (whisper.volumeMounts ?? []).map((m) => m.mountPath).sort(), + ['/tmp', '/work'], + 'whisper mounts more than the work volume and /tmp', + ); + assert.equal(whisper.securityContext.runAsNonRoot, true); + assert.equal(whisper.securityContext.runAsUser, 1000); + assert.equal(whisper.securityContext.readOnlyRootFilesystem, true); + assert.equal(whisper.securityContext.allowPrivilegeEscalation, false); + assert.deepEqual(whisper.securityContext.capabilities.drop, ['ALL']); + assert.deepEqual(whisper.securityContext.seccompProfile, { type: 'RuntimeDefault' }); + for (const secret of ['apicore-backend-secret', ANTHROPIC_SECRET, TELEGRAM_SECRET]) { + assert.deepEqual( + secretReferences(pod, secret).filter((r) => r.container === 'whisper'), + [], + `whisper references secret ${secret}`, + ); + } +}); + +test('the voice pod matches the jobs network policy (egress only postgres and dns)', () => { + const pod = voicePodOf(target); + const postgres = postgresOf(target); + assert.equal(pod.labels.app, 'apicore-job', 'the pod template must carry app=apicore-job for apicore-jobs'); + assert.equal(pod.labels['apicore.io/job'], 'voice-transcribe'); + assert.equal(pod.labels['db-client'], 'true'); + assert.ok(target.canConnect(pod, postgres, 5432), 'voice pod cannot reach postgres'); + assert.ok(!target.canConnect(pod, postgres, 22), 'voice pod reaches postgres on another port'); + for (const protocol of ['UDP', 'TCP']) { + assert.ok(target.canConnect(pod, KUBE_DNS, 53, protocol), `voice pod cannot resolve names (${protocol})`); + } + for (const address of ['1.1.1.1', '10.43.0.1', '169.254.169.254']) { + for (const port of [80, 443]) { + assert.ok(!target.canConnect(pod, ip(address), port), `voice pod reaches ${address}:${port}`); + } + } + for (const other of target.pods.filter((p) => p !== postgres && p !== pod)) { + assert.ok(!target.canConnect(pod, other, 3001) && !target.canConnect(pod, other, 3000), `voice pod reaches ${other.owner}`); + } + for (const policy of target.policiesFor(pod, 'Ingress')) { + assert.equal((policy.spec.ingress ?? []).length, 0, `${policy.metadata.name} opens ingress to the voice pod`); + } +}); + +test('work volume is a memory emptyDir of at most 32Mi, mounted in all three containers', () => { + const pod = voicePodOf(target); + const work = (pod.spec.volumes ?? []).find((v) => v.name === 'work'); + assert.ok(work?.emptyDir, 'work volume missing or not an emptyDir'); + assert.equal(work.emptyDir.medium, 'Memory', 'work volume is not memory-backed'); + assert.equal(work.emptyDir.sizeLimit, '32Mi'); + for (const name of ['fetch', 'whisper', 'ingest']) { + const mount = (voiceContainer(pod, name).volumeMounts ?? []).find((m) => m.name === 'work'); + assert.equal(mount?.mountPath, '/work', `${name} does not mount work on /work`); + assert.ok(!mount.readOnly, `${name} mounts work read-only`); + } + // fetch, whisper and ingest share the files (0700/0600): one uid, one fsGroup. + assert.equal(pod.spec.securityContext.runAsUser, 1000); + assert.equal(pod.spec.securityContext.fsGroup, 1000); + for (const name of ['fetch', 'whisper', 'ingest']) { + const own = voiceContainer(pod, name).securityContext.runAsUser; + assert.ok(own === undefined || own === 1000, `${name} runs as another uid`); + } +}); + +test('voice resources match spec 11.1', () => { + const pod = voicePodOf(target); + assert.deepEqual(voiceContainer(pod, 'fetch').resources, res('50m', '500m', '192Mi', '512Mi')); + assert.deepEqual(voiceContainer(pod, 'ingest').resources, res('50m', '500m', '192Mi', '512Mi')); + assert.deepEqual(voiceContainer(pod, 'whisper').resources, res('50m', '2000m', '1280Mi', '1280Mi')); +}); + +test('the voice CronJob is not matched by the shared job patch and has exactly fetch, whisper, ingest', () => { + const cronJob = voiceCronJob(target); + assert.notEqual(cronJob.metadata.labels?.app, 'apicore-job', 'the shared job patch would add a container `job`'); + const pod = voicePodOf(target); + assert.deepEqual(pod.initContainers.map((c) => c.name), ['fetch', 'whisper']); + assert.deepEqual(pod.containers.map((c) => c.name), ['ingest']); + assert.ok(![...pod.initContainers, ...pod.containers].some((c) => c.name === 'job'), 'a container `job` was patched in'); + // Everything the shared patch would have set is spelled out in the manifest. + const { spec } = cronJob; + assert.equal(spec.schedule, '*/2 * * * *'); + assert.equal(spec.timeZone, 'Europe/Berlin'); + assert.equal(spec.concurrencyPolicy, 'Forbid'); + assert.equal(spec.startingDeadlineSeconds, 120); + assert.equal(spec.suspend ?? false, false); + assert.equal(spec.jobTemplate.spec.activeDeadlineSeconds, 1500); + assert.equal(spec.jobTemplate.spec.backoffLimit, 0); + assert.equal(spec.jobTemplate.spec.ttlSecondsAfterFinished, 86400); + assert.equal(pod.spec.restartPolicy, 'Never'); + assert.equal(pod.spec.securityContext.runAsNonRoot, true); + assert.deepEqual(pod.spec.securityContext.seccompProfile, { type: 'RuntimeDefault' }); + // fetch and ingest are the backend image in the job role; both phases are exact commands. + const fetch = voiceContainer(pod, 'fetch'); + const ingest = voiceContainer(pod, 'ingest'); + assert.deepEqual(fetch.command, ['node', 'dist/cli.js', 'voice-transcribe', '--phase=fetch']); + assert.deepEqual(ingest.command, ['node', 'dist/cli.js', 'voice-transcribe', '--phase=ingest']); + assert.ok(KNOWN_JOBS.includes(fetch.command[2])); + for (const c of [fetch, ingest]) { + assert.equal(envOf(c).APICORE_ROLE, 'job', `${c.name}: APICORE_ROLE`); + assert.equal(c.image.split('@')[0], 'git.encry.de/nick.rosenloecher/apicore-backend'); + assert.ok((c.envFrom ?? []).some((f) => f.secretRef?.name === 'apicore-backend-secret'), `${c.name}: backend secret`); + } +}); + +// The Claude extraction is on hold (no worker patch yet). Until it lands, no pod +// at all may carry the key; once the worker patch exists the worker is exempt. +for (const [label, model] of [['target', target], ...stages]) { + test(`${label}: ${ANTHROPIC_KEY} appears only in the worker (no other pod references it)`, () => { + for (const pod of model.pods) { + if (pod.labels.app === 'apicore-worker') continue; + for (const c of [...pod.containers, ...pod.initContainers]) { + assert.ok(!(c.env ?? []).some((e) => e.name === ANTHROPIC_KEY), `${pod.owner}/${c.name} defines ${ANTHROPIC_KEY}`); + } + assert.deepEqual(secretReferences(pod, ANTHROPIC_SECRET), [], `${pod.owner} must not use secret ${ANTHROPIC_SECRET}`); + } + for (const doc of model.docs.filter((d) => d.kind === 'ConfigMap')) { + assert.ok(!(ANTHROPIC_KEY in (doc.data ?? {})), `${doc.metadata.name} holds ${ANTHROPIC_KEY}`); + } + }); +} + +test('the voice settings sit in the shared ConfigMap with the values the backend accepts', () => { + const configMap = target.docs.find((d) => d.kind === 'ConfigMap' && d.metadata.name === 'apicore-backend-config'); + assert.equal(configMap.data.VOICE_RECORDING_ENABLED, 'false'); + assert.equal(configMap.data.VOICE_MODEL, 'claude-sonnet-5-5'); + assert.equal(configMap.data.VOICE_DAILY_EXTRACTIONS, '60'); +}); + +test('apicore-whisper is pinned by digest in the overlay', () => { + const overlay = readFileSync(new URL('../overlays/encry/kustomization.yaml', import.meta.url), 'utf8'); + assert.match(overlay, new RegExp(`name: ${WHISPER_IMAGE}\\s+digest: sha256:[0-9a-f]{64}`), 'no digest entry for apicore-whisper'); + const whisper = voiceContainer(voicePodOf(target), 'whisper'); + assert.match(whisper.image, new RegExp(`^${WHISPER_IMAGE}@sha256:[0-9a-f]{64}$`)); + assert.doesNotMatch(whisper.image, /:latest/); +}); + +test('no volume except work is mounted by both whisper and fetch/ingest; /tmp volumes are size-limited', () => { + const pod = voicePodOf(target); + const mounted = (name) => new Set((voiceContainer(pod, name).volumeMounts ?? []).map((m) => m.name)); + const whisper = mounted('whisper'); + for (const other of ['fetch', 'ingest']) { + const shared = [...mounted(other)].filter((v) => whisper.has(v)); + assert.deepEqual(shared, ['work'], `whisper and ${other} share volumes other than work: ${shared}`); + } + const tmpOf = (name) => voiceContainer(pod, name).volumeMounts.find((m) => m.mountPath === '/tmp').name; + assert.notEqual(tmpOf('whisper'), tmpOf('ingest'), 'whisper and ingest share /tmp'); + for (const volume of pod.spec.volumes) { + assert.ok(volume.emptyDir?.sizeLimit, `volume ${volume.name} has no sizeLimit`); + } + assert.equal(pod.spec.volumes.find((v) => v.name === tmpOf('ingest')).emptyDir.sizeLimit, '64Mi'); +}); diff --git a/deploy/k8s/tests/render-stages.mjs b/deploy/k8s/tests/render-stages.mjs index 1996daa..9009a54 100644 --- a/deploy/k8s/tests/render-stages.mjs +++ b/deploy/k8s/tests/render-stages.mjs @@ -1,4 +1,4 @@ -// Renders every rollout stage of overlays/encry (Runbook steps 1, 2, 3 and 5) +// Renders every rollout stage of overlays/encry (Runbook steps 1, 2, 3, 5 and the voice component) // so manifest-policy.test.mjs can check that each intermediate state keeps // the application reachable. A stage is the committed overlay with only the // components released up to that step; the overlay in the repo is not @@ -15,12 +15,13 @@ import { fileURLToPath } from 'node:url'; const K8S_DIR = resolve(dirname(fileURLToPath(import.meta.url)), '..'); const OVERLAY_FILE = join('overlays', 'encry', 'kustomization.yaml'); -const ROLLOUT_ORDER = ['network-policies', 'pod-hardening', 'backend-roles', 'api-hpa']; +const ROLLOUT_ORDER = ['network-policies', 'pod-hardening', 'backend-roles', 'api-hpa', 'voice']; const STAGES = { 'step1-ist-stand': [], 'step2-network-policies': ROLLOUT_ORDER.slice(0, 1), 'step3-roles-and-hardening': ROLLOUT_ORDER.slice(0, 3), - 'step5-hpa': ROLLOUT_ORDER, + 'step5-hpa': ROLLOUT_ORDER.slice(0, 4), + 'step6-voice': ROLLOUT_ORDER, }; const COMPONENT_LINE = /^\s+-\s+\.\.\/\.\.\/components\/([\w-]+)/;