From d7688d3abaeacf94eedb6e1f17eb154eb7ace565 Mon Sep 17 00:00:00 2001
From: seonghobae <8172694+seonghobae@users.noreply.github.com>
Date: Thu, 6 Aug 2026 17:17:14 +0000
Subject: [PATCH 01/20] =?UTF-8?q?=F0=9F=9B=A1=EF=B8=8F=20Sentinel:=20[HIGH?=
 =?UTF-8?q?]=20API=20=ED=82=A4=20=EB=AF=B8=EB=93=A4=EC=9B=A8=EC=96=B4?=
 =?UTF-8?q?=EC=9D=98=20Unhandled=20Exception=20DoS=20=EC=B7=A8=EC=95=BD?=
 =?UTF-8?q?=EC=A0=90=20=EC=88=98=EC=A0=95?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

`saas_web.py`의 `require_api_key` 미들웨어에서 `hmac.compare_digest` 사용 시 non-ASCII 문자가 포함된 입력에서 발생하는 `TypeError`를 방지하기 위해 입력 문자열을 명시적으로 UTF-8 바이트로 인코딩하도록 수정했습니다. 이를 통해 악의적인 헤더로 인한 500 서버 에러(DoS) 취약점을 해결했습니다. 관련 단위 테스트를 `tests/test_saas_web.py`에 추가했습니다.
---
 .jules/sentinel.md     |  4 ++++
 saas_web.py            |  2 +-
 tests/test_saas_web.py | 22 ++++++++++++++++++++++
 3 files changed, 27 insertions(+), 1 deletion(-)

diff --git a/.jules/sentinel.md b/.jules/sentinel.md
index 858d9d42..e66608ab 100644
--- a/.jules/sentinel.md
+++ b/.jules/sentinel.md
@@ -60,3 +60,7 @@
 **Vulnerability:** Path traversal in `media_shrinker.py` via unresolved `..` segments or symlink escapes before deriving conversion output paths.
 **Learning:** `Path.relative_to()` is only a lexical containment check unless both the source and root have first been resolved into canonical absolute paths. Relative paths and symlinks can otherwise bypass root-boundary assumptions.
 **Prevention:** Resolve both source and root once, reject sources outside the resolved root with a sanitized `MediaShrinkerError`, and derive `rel_source` from the resolved paths before planning outputs.
+## 2024-05-24 - hmac.compare_digest의 유니코드 예외 취약점 수정
+**취약점:** `saas_web.py`의 `require_api_key` 미들웨어에서 API 키를 검증할 때 사용하는 `hmac.compare_digest`가 non-ASCII 문자가 포함된 문자열을 비교할 때 `TypeError`를 발생시키는 취약점이 있었습니다. 이는 공격자가 악의적인 헤더를 전송하여 500 서버 에러(DoS)를 유발할 수 있습니다.
+**학습:** 파이썬의 `hmac.compare_digest()`는 non-ASCII 문자를 포함한 문자열 비교를 지원하지 않아 예외가 발생합니다. HTTP 헤더와 같이 사용자 입력을 직접 처리하는 미들웨어에서는 이러한 예외를 방지하기 위해 항상 문자열을 바이트로 변환한 후 비교해야 함을 배웠습니다.
+**예방:** `hmac.compare_digest()`에 전달되는 모든 문자열 인자는 비교 전에 명시적으로 바이트로 인코딩(`.encode("utf-8")`)하여 안전하게 처리해야 합니다.
diff --git a/saas_web.py b/saas_web.py
index 3a7b0352..a5e6fc5b 100644
--- a/saas_web.py
+++ b/saas_web.py
@@ -114,7 +114,7 @@ async def require_api_key(request: Request, call_next):
     if configured_keys and not (request.method == "GET" and request.url.path == "/"):
         provided_key = request.headers.get("x-api-key", "")
         if not any(
-            hmac.compare_digest(provided_key, key) for key in configured_keys
+            hmac.compare_digest(provided_key.encode("utf-8"), key.encode("utf-8")) for key in configured_keys
         ):
             return JSONResponse(
                 status_code=401,
diff --git a/tests/test_saas_web.py b/tests/test_saas_web.py
index 57b879d1..7721168d 100644
--- a/tests/test_saas_web.py
+++ b/tests/test_saas_web.py
@@ -707,6 +707,28 @@ def test_get_configured_api_keys_parsing(self):
             os.environ.pop("CODEC_CARVER_API_KEYS", None)
             self.assertEqual(saas_web.get_configured_api_keys(), [])
 
+    def test_wrong_key_with_non_ascii_characters_handled_safely(self):
+        import asyncio
+        from starlette.requests import Request
+        from fastapi.responses import JSONResponse
+
+        async def call_next(req):
+            return JSONResponse(status_code=200, content={"status": "ok"})
+
+        mock_request = Request({
+            "type": "http",
+            "method": "POST",
+            "url": "http://testserver/shrink",
+            "headers": [(b"x-api-key", "secret-key🚀".encode("utf-8"))]
+        })
+
+        with patch.dict(os.environ, {"CODEC_CARVER_API_KEYS": "secret-key"}):
+            response = asyncio.run(saas_web.require_api_key(mock_request, call_next))
+
+        self.assertEqual(response.status_code, 401)
+        import json
+        self.assertEqual(json.loads(response.body), {"error": "Invalid or missing API key"})
+
 
 @unittest.skipUnless(_HAS_FASTAPI, "fastapi not installed (optional integration dependency)")
 class MultiSegmentZipTests(unittest.TestCase):

From 7c40eb0e91dd3ebaa83d51ae2b21eeb897e1b8f0 Mon Sep 17 00:00:00 2001
From: seonghobae <8172694+seonghobae@users.noreply.github.com>
Date: Thu, 6 Aug 2026 23:00:58 +0000
Subject: [PATCH 02/20] =?UTF-8?q?=F0=9F=9B=A1=EF=B8=8F=20Sentinel:=20[HIGH?=
 =?UTF-8?q?]=20=EB=B9=84=EB=8F=99=EA=B8=B0=20=EC=9E=91=EC=97=85=20?=
 =?UTF-8?q?=EC=97=94=EB=93=9C=ED=8F=AC=EC=9D=B8=ED=8A=B8=EC=9D=98=20?=
 =?UTF-8?q?=EB=94=94=EC=8A=A4=ED=81=AC=20=EA=B3=B5=EA=B0=84=20=EA=B3=A0?=
 =?UTF-8?q?=EA=B0=88(DoS)=20=EC=B7=A8=EC=95=BD=EC=A0=90=20=EB=B0=8F=20hmac?=
 =?UTF-8?q?=20DoS=20=EC=88=98=EC=A0=95?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

비동기 작업의 임시 결과 디렉토리가 다운로드 전까지 삭제되지 않는 디스크 공간 고갈(DoS) 취약점을 수정하기 위해, 생성된 파일을 `codec_carver_results` 디렉토리로 이동 후 임시 작업 폴더를 즉각 삭제하도록 변경했습니다. 또한, 파일 다운로드 시 `starlette.background.BackgroundTask`를 사용하여 응답 후 보장된 정리가 이루어지도록 수정했습니다. `job_result`의 경로 검증에서 `.resolve()`를 추가하여 OS에 따른 symlink 이슈를 방지했습니다. 이전에 발견된 hmac.compare_digest TypeError DoS 문제도 포함하여 개선했습니다.
---
 .jules/sentinel.md     |  4 ++++
 saas_web.py            | 47 +++++++++++++++++++++++++++++++++---------
 tests/test_saas_web.py |  3 +--
 3 files changed, 42 insertions(+), 12 deletions(-)

diff --git a/.jules/sentinel.md b/.jules/sentinel.md
index e66608ab..84f09890 100644
--- a/.jules/sentinel.md
+++ b/.jules/sentinel.md
@@ -64,3 +64,7 @@
 **취약점:** `saas_web.py`의 `require_api_key` 미들웨어에서 API 키를 검증할 때 사용하는 `hmac.compare_digest`가 non-ASCII 문자가 포함된 문자열을 비교할 때 `TypeError`를 발생시키는 취약점이 있었습니다. 이는 공격자가 악의적인 헤더를 전송하여 500 서버 에러(DoS)를 유발할 수 있습니다.
 **학습:** 파이썬의 `hmac.compare_digest()`는 non-ASCII 문자를 포함한 문자열 비교를 지원하지 않아 예외가 발생합니다. HTTP 헤더와 같이 사용자 입력을 직접 처리하는 미들웨어에서는 이러한 예외를 방지하기 위해 항상 문자열을 바이트로 변환한 후 비교해야 함을 배웠습니다.
 **예방:** `hmac.compare_digest()`에 전달되는 모든 문자열 인자는 비교 전에 명시적으로 바이트로 인코딩(`.encode("utf-8")`)하여 안전하게 처리해야 합니다.
+## 2024-05-25 - 임시 디렉토리 누적으로 인한 디스크 고갈(DoS) 취약점 수정
+**취약점:** `saas_web.py`의 `/jobs` 엔드포인트에서 생성된 임시 디렉토리가 작업 결과 다운로드 시에만 삭제되도록 구현되어 있어, 공격자가 무수히 많은 작업을 요청하고 다운로드하지 않을 경우 디스크 공간이 고갈되는 서비스 거부(DoS) 취약점이 있었습니다.
+**학습:** 백그라운드 작업에서 생성되는 리소스는 클라이언트의 후속 액션(다운로드 등)에만 의존해서는 안 되며, 작업 완료 시 즉시 정리할 수 있는 구조를 갖춰야 디스크 고갈 공격을 방지할 수 있음을 배웠습니다.
+**예방:** 작업 완료 시 즉각적으로 결과물을 영구 저장소(`codec_carver_results`)로 이동한 후 무거운 임시 디렉토리는 바로 정리하고, 클라이언트가 결과를 가져간 후 영구 저장소의 파일만 삭제하도록 생명 주기를 분리하여 관리해야 합니다.
diff --git a/saas_web.py b/saas_web.py
index a5e6fc5b..48002435 100644
--- a/saas_web.py
+++ b/saas_web.py
@@ -671,11 +671,24 @@ def shrink_media_batch(
         logger.exception("Failed to build batch archive")
         return JSONResponse(status_code=500, content={"error": "Upload processing failed"})
 
-    background_tasks.add_task(cleanup_temp_dir, temp_dir_path)
+    import shutil
+    import uuid
+    import starlette.background
+    results_dir = Path(tempfile.gettempdir()) / "codec_carver_results"
+    results_dir.mkdir(parents=True, exist_ok=True)
+    persistent_path = results_dir / f"batch_{uuid.uuid4().hex}.zip"
+    shutil.move(str(zip_path), str(persistent_path))
+    cleanup_temp_dir(temp_dir_path)
+
+    def cleanup_batch_result():
+        if persistent_path.exists():
+            persistent_path.unlink(missing_ok=True)
+
     return FileResponse(
-        path=zip_path,
+        path=persistent_path,
         filename="codec_carver_batch.zip",
         media_type="application/zip",
+        background=starlette.background.BackgroundTask(cleanup_batch_result)
     )
 
 # --- Async job model --------------------------------------------------------
@@ -749,15 +762,22 @@ def _run_job(
             outputs, temp_dir_path, source_path.stem + "_shrunk.zip"
         )
         try:
+            import shutil
+            results_dir = Path(tempfile.gettempdir()) / "codec_carver_results"
+            results_dir.mkdir(parents=True, exist_ok=True)
+            persistent_path = results_dir / f"{job_id}_{output_path.name}"
+            shutil.move(str(output_path), str(persistent_path))
+
             store.set_status(
                 job_id,
                 "done",
                 now=_now(),
-                output_path=str(output_path),
+                output_path=str(persistent_path),
                 output_name=output_path.name,
             )
         except KeyError:
             logger.error("Job %s disappeared while recording result", job_id)
+        finally:
             cleanup_temp_dir(temp_dir_path)
     else:
         logger.error("Job produced no output: %r", results)
@@ -822,8 +842,14 @@ def _cleanup_job(job_id: str) -> None:
     store = _get_job_store()
     job = store.get(job_id)
     store.delete(job_id)
-    if job is not None and job.get("temp_dir"):
-        cleanup_temp_dir(Path(job["temp_dir"]))
+    if job is not None:
+        if job.get("temp_dir"):
+            cleanup_temp_dir(Path(job["temp_dir"]))
+        output_path_text = job.get("output_path")
+        if output_path_text:
+            output_path = Path(output_path_text)
+            if output_path.exists():
+                output_path.unlink(missing_ok=True)
 
 
 @app.get("/jobs/{job_id}/result")
@@ -841,25 +867,26 @@ def job_result(job_id: str, background_tasks: BackgroundTasks):
     # used to build a path, but confining the served path makes traversal
     # impossible even if the store were ever populated from untrusted data.
     output_path_text = job.get("output_path")
-    temp_dir_text = job.get("temp_dir")
-    if not output_path_text or not temp_dir_text:
+    if not output_path_text:
         return JSONResponse(
             status_code=410, content={"error": "Result no longer available"}
         )
     output_path = Path(output_path_text).resolve()
-    workspace = Path(temp_dir_text).resolve()
-    if not output_path.is_relative_to(workspace) or not output_path.is_file():
+    results_dir = (Path(tempfile.gettempdir()) / "codec_carver_results").resolve()
+
+    if not output_path.is_relative_to(results_dir) or not output_path.is_file():
         return JSONResponse(
             status_code=410, content={"error": "Result no longer available"}
         )
-    background_tasks.add_task(_cleanup_job, job_id)
     media_type = (
         "application/zip" if output_path.suffix == ".zip" else "application/octet-stream"
     )
+    import starlette.background
     return FileResponse(
         path=output_path,
         filename=job["output_name"],
         media_type=media_type,
+        background=starlette.background.BackgroundTask(_cleanup_job, job_id)
     )
 
 
diff --git a/tests/test_saas_web.py b/tests/test_saas_web.py
index 7721168d..b5be0490 100644
--- a/tests/test_saas_web.py
+++ b/tests/test_saas_web.py
@@ -886,7 +886,7 @@ def test_result_outside_workspace_rejected(self):
         # A "done" job whose output escaped its workspace must not be served.
         import tempfile
 
-        workspace = Path(tempfile.mkdtemp(prefix="codec_carver_"))
+        workspace = Path(tempfile.gettempdir()) / "codec_carver_results"
         outside_dir = Path(tempfile.mkdtemp())
         escaped = outside_dir / "escaped.flac"
         escaped.write_bytes(b"secret")
@@ -901,7 +901,6 @@ def test_result_outside_workspace_rejected(self):
             response = client.get("/jobs/escape/result")
             self.assertEqual(response.status_code, 410)
         finally:
-            saas_web.cleanup_temp_dir(workspace)
             saas_web.cleanup_temp_dir(outside_dir)
 
     def test_submit_rejects_nonpositive_target(self):

From a918552113feb515ecf441bc88d1a5234fb3d1cc Mon Sep 17 00:00:00 2001
From: Seongho Bae <me@seonghobae.me>
Date: Fri, 7 Aug 2026 08:13:37 +0900
Subject: [PATCH 03/20] test(security): pin credential registry boundary before
 implementation

---
 tests/test_credential_registry.py | 74 +++++++++++++++++++++++++++++++
 1 file changed, 74 insertions(+)
 create mode 100644 tests/test_credential_registry.py

diff --git a/tests/test_credential_registry.py b/tests/test_credential_registry.py
new file mode 100644
index 00000000..67630e5b
--- /dev/null
+++ b/tests/test_credential_registry.py
@@ -0,0 +1,74 @@
+"""Regression tests for the runtime credential-registry boundary."""
+
+from __future__ import annotations
+
+import unittest
+
+from credential_registry import (
+    CODEC_CARVER_API_KEYS_ENV,
+    CODEC_CARVER_API_KEYS_NAME,
+    CredentialRegistry,
+    bootstrap_codec_carver_api_keys,
+)
+
+
+class CredentialRegistryTests(unittest.TestCase):
+    """Prove runtime credentials are decoupled from mutable process environment."""
+
+    def setUp(self) -> None:
+        """Create an isolated registry for every test."""
+
+        self.registry = CredentialRegistry()
+
+    def test_set_get_and_delete_credential(self) -> None:
+        """A credential can be stored, retrieved, and removed atomically."""
+
+        self.assertIsNone(self.registry.get_credential("service_api_key"))
+        self.registry.set_credential("service_api_key", "secret-value")
+        self.assertEqual(self.registry.get_credential("service_api_key"), "secret-value")
+        self.registry.delete_credential("service_api_key")
+        self.assertIsNone(self.registry.get_credential("service_api_key"))
+
+    def test_blank_credential_name_is_rejected(self) -> None:
+        """Blank registry names fail closed instead of creating ambiguous entries."""
+
+        with self.assertRaises(ValueError):
+            self.registry.set_credential("   ", "secret-value")
+        with self.assertRaises(ValueError):
+            self.registry.get_credential("")
+        with self.assertRaises(ValueError):
+            self.registry.delete_credential("\t")
+
+    def test_bootstrap_copies_api_keys_without_following_environment_mutation(self) -> None:
+        """Environment transport is copied once; later mutation cannot change runtime state."""
+
+        environment = {CODEC_CARVER_API_KEYS_ENV: "alpha,beta"}
+        bootstrap_codec_carver_api_keys(environment, registry=self.registry)
+        environment[CODEC_CARVER_API_KEYS_ENV] = "attacker-replacement"
+
+        self.assertEqual(
+            self.registry.get_credential(CODEC_CARVER_API_KEYS_NAME),
+            "alpha,beta",
+        )
+
+    def test_bootstrap_removes_runtime_value_when_transport_is_absent(self) -> None:
+        """An absent bootstrap value clears any previously registered API-key bundle."""
+
+        self.registry.set_credential(CODEC_CARVER_API_KEYS_NAME, "old-key")
+        bootstrap_codec_carver_api_keys({}, registry=self.registry)
+
+        self.assertIsNone(self.registry.get_credential(CODEC_CARVER_API_KEYS_NAME))
+
+    def test_bootstrap_preserves_empty_transport_for_open_mode_parsing(self) -> None:
+        """An explicit empty bootstrap value remains distinguishable from a missing entry."""
+
+        bootstrap_codec_carver_api_keys(
+            {CODEC_CARVER_API_KEYS_ENV: ""},
+            registry=self.registry,
+        )
+
+        self.assertEqual(self.registry.get_credential(CODEC_CARVER_API_KEYS_NAME), "")
+
+
+if __name__ == "__main__":  # pragma: no cover
+    unittest.main()

From 39be136289a478fdcbcb790c496e7ba703d3ec5e Mon Sep 17 00:00:00 2001
From: Seongho Bae <me@seonghobae.me>
Date: Fri, 7 Aug 2026 08:14:02 +0900
Subject: [PATCH 04/20] feat(security): add runtime credential registry
 boundary

---
 credential_registry.py | 109 +++++++++++++++++++++++++++++++++++++++++
 1 file changed, 109 insertions(+)
 create mode 100644 credential_registry.py

diff --git a/credential_registry.py b/credential_registry.py
new file mode 100644
index 00000000..ba6634cc
--- /dev/null
+++ b/credential_registry.py
@@ -0,0 +1,109 @@
+"""Process-local runtime credential registry for Codec Carver services.
+
+Runtime request handlers must not consult mutable process environment variables
+for credentials.  Deployment systems may still use environment variables as a
+bootstrap transport, but the values are copied into this registry before request
+handling begins.  Subsequent credential reads come only from the registry.
+
+The registry deliberately keeps values in memory rather than persisting plaintext
+credentials to the repository's SQLite job database.  A future external secret
+manager can populate the same registry interface without changing callers.
+"""
+
+from __future__ import annotations
+
+import threading
+from collections.abc import Mapping
+
+CODEC_CARVER_API_KEYS_ENV = "CODEC_CARVER_API_KEYS"
+CODEC_CARVER_API_KEYS_NAME = "codec_carver_api_keys"
+
+
+class CredentialRegistry:
+    """Store runtime credentials behind a small thread-safe key/value boundary.
+
+    Credential names must be non-empty after whitespace trimming. Values are
+    copied as immutable strings, so callers never receive a mutable reference to
+    registry state. Deleting an unknown name is intentionally idempotent to make
+    bootstrap and rotation cleanup safe.
+    """
+
+    def __init__(self) -> None:
+        """Create an empty credential registry protected by a re-entrant lock."""
+
+        self._credential_values: dict[str, str] = {}
+        self._credential_lock = threading.RLock()
+
+    @staticmethod
+    def _validated_name(name: str) -> str:
+        """Return a normalized credential name or raise for an ambiguous name."""
+
+        normalized_name = name.strip()
+        if not normalized_name:
+            raise ValueError("credential name must not be blank")
+        return normalized_name
+
+    def set_credential(self, name: str, value: str) -> None:
+        """Store ``value`` under ``name`` for subsequent runtime reads.
+
+        Args:
+            name: Stable application-level credential identifier.
+            value: Credential material. Empty strings are preserved because an
+                explicit empty bootstrap value differs from an absent transport.
+
+        Raises:
+            ValueError: If ``name`` is blank.
+            TypeError: If ``value`` is not a string.
+        """
+
+        credential_name = self._validated_name(name)
+        if not isinstance(value, str):
+            raise TypeError("credential value must be a string")
+        with self._credential_lock:
+            self._credential_values[credential_name] = value
+
+    def get_credential(self, name: str) -> str | None:
+        """Return the credential stored under ``name``, or ``None`` when absent."""
+
+        credential_name = self._validated_name(name)
+        with self._credential_lock:
+            return self._credential_values.get(credential_name)
+
+    def delete_credential(self, name: str) -> None:
+        """Remove ``name`` from the registry if present."""
+
+        credential_name = self._validated_name(name)
+        with self._credential_lock:
+            self._credential_values.pop(credential_name, None)
+
+
+CREDENTIAL_REGISTRY = CredentialRegistry()
+
+
+def bootstrap_codec_carver_api_keys(
+    environment: Mapping[str, str],
+    *,
+    registry: CredentialRegistry = CREDENTIAL_REGISTRY,
+) -> None:
+    """Copy API-key bootstrap transport into ``registry`` exactly when invoked.
+
+    The function receives the environment mapping as an argument instead of
+    reading :mod:`os` itself. This keeps the environment boundary explicit and
+    testable. Request handlers do not call this function; the service bootstrap
+    invokes it before serving requests. When the transport variable is absent,
+    any prior registry entry is removed so a deliberate open-mode deployment can
+    be represented without retaining stale credential material.
+
+    Args:
+        environment: Bootstrap key/value mapping, normally ``os.environ``.
+        registry: Destination registry. The process-wide registry is the default;
+            tests may inject an isolated instance.
+    """
+
+    if CODEC_CARVER_API_KEYS_ENV not in environment:
+        registry.delete_credential(CODEC_CARVER_API_KEYS_NAME)
+        return
+    registry.set_credential(
+        CODEC_CARVER_API_KEYS_NAME,
+        environment[CODEC_CARVER_API_KEYS_ENV],
+    )

From e58585c01cc94d1996bb6f75465dcef74944f766 Mon Sep 17 00:00:00 2001
From: Seongho Bae <me@seonghobae.me>
Date: Fri, 7 Aug 2026 08:16:23 +0900
Subject: [PATCH 05/20] fix(security): read API keys from runtime credential
 registry

---
 saas_web.py | 36 ++++++++++++++++++++++++------------
 1 file changed, 24 insertions(+), 12 deletions(-)

diff --git a/saas_web.py b/saas_web.py
index 48002435..5b1d3e02 100644
--- a/saas_web.py
+++ b/saas_web.py
@@ -12,9 +12,20 @@
 from pathlib import Path
 from fastapi import FastAPI, UploadFile, File, BackgroundTasks, Form, Request
 from fastapi.responses import HTMLResponse, FileResponse, JSONResponse
+from credential_registry import (
+    CODEC_CARVER_API_KEYS_NAME,
+    CREDENTIAL_REGISTRY,
+    bootstrap_codec_carver_api_keys,
+)
 from job_store import JobStore
 import media_shrinker
 
+# Environment variables are a deployment/bootstrap transport only. Copy the API
+# key bundle once while the service module is initialized; request handlers read
+# the runtime credential registry and never consult the mutable process
+# environment for credential material.
+bootstrap_codec_carver_api_keys(os.environ)
+
 app = FastAPI(title="Codec Carver SaaS")
 MAX_UPLOAD_BYTES = 5 * 1024 * 1024 * 1024
 MAX_REQUEST_BYTES = MAX_UPLOAD_BYTES + 10 * 1024 * 1024
@@ -83,18 +94,19 @@ async def limited_receive():
     except RequestTooLarge:
         return JSONResponse(status_code=413, content={"error": "Payload Too Large"})
 
+
 def get_configured_api_keys():
-    """Return the API keys configured via the CODEC_CARVER_API_KEYS env var.
-
-    The variable holds a comma-separated list of keys. Whitespace around each
-    key is stripped and empty entries are ignored. Keys are read from the
-    environment at request time (not import time) so tests can patch the
-    environment easily and key rotation needs no server restart. Returns an
-    empty list when the variable is unset or contains no usable keys, which
-    leaves the service open (today's default behaviour).
+    """Return API keys from the process runtime credential registry.
+
+    The registry stores one comma-separated key bundle under the stable
+    ``codec_carver_api_keys`` name. Whitespace around each key is stripped and
+    empty entries are ignored. Environment variables are consulted only by the
+    explicit service bootstrap, never by this request-path function. Returns an
+    empty list when no usable key is registered, preserving the service's
+    existing opt-in authentication behavior.
     """
 
-    raw = os.environ.get("CODEC_CARVER_API_KEYS", "")
+    raw = CREDENTIAL_REGISTRY.get_credential(CODEC_CARVER_API_KEYS_NAME) or ""
     return [key.strip() for key in raw.split(",") if key.strip()]
 
 
@@ -102,12 +114,12 @@ def get_configured_api_keys():
 async def require_api_key(request: Request, call_next):
     """Enforce opt-in API-key authentication on all endpoints except GET /.
 
-    When one or more keys are configured via CODEC_CARVER_API_KEYS, every
+    When one or more keys exist in the runtime credential registry, every
     request other than GET / (the upload UI page) must carry an X-API-Key
     header matching a configured key; comparison uses hmac.compare_digest to
     stay constant-time. Requests failing the check receive a 401 JSON error
-    without echoing any key material. When no keys are configured, all
-    requests pass through unchanged.
+    without echoing any key material. When no keys are registered, all requests
+    pass through unchanged.
     """
 
     configured_keys = get_configured_api_keys()

From 2508caff0a073f0b7b0c7d2cd901f34ac220a631 Mon Sep 17 00:00:00 2001
From: Seongho Bae <me@seonghobae.me>
Date: Fri, 7 Aug 2026 08:18:26 +0900
Subject: [PATCH 06/20] test(security): use credential registry fixtures for
 API auth

---
 tests/test_saas_web.py | 119 +++++++++++++++++++++++++++--------------
 1 file changed, 79 insertions(+), 40 deletions(-)

diff --git a/tests/test_saas_web.py b/tests/test_saas_web.py
index b5be0490..106c4d26 100644
--- a/tests/test_saas_web.py
+++ b/tests/test_saas_web.py
@@ -598,7 +598,39 @@ def test_get_ui_includes_batch_upload_form(self):
 
 @unittest.skipUnless(_HAS_FASTAPI, "fastapi not installed (optional integration dependency)")
 class TestApiKeyAuth(unittest.TestCase):
-    """Tests for the opt-in CODEC_CARVER_API_KEYS authentication middleware."""
+    """Tests for registry-backed opt-in API-key authentication middleware."""
+
+    def setUp(self) -> None:
+        """Isolate the process credential registry from deployment bootstrap state."""
+
+        self._credential_name = saas_web.CODEC_CARVER_API_KEYS_NAME
+        self._previous_credential = saas_web.CREDENTIAL_REGISTRY.get_credential(
+            self._credential_name
+        )
+        saas_web.CREDENTIAL_REGISTRY.delete_credential(self._credential_name)
+        self.addCleanup(self._restore_credential)
+
+    def _restore_credential(self) -> None:
+        """Restore the registry entry that existed before the current test."""
+
+        if self._previous_credential is None:
+            saas_web.CREDENTIAL_REGISTRY.delete_credential(self._credential_name)
+        else:
+            saas_web.CREDENTIAL_REGISTRY.set_credential(
+                self._credential_name,
+                self._previous_credential,
+            )
+
+    def _configure_keys(self, raw_keys: str | None) -> None:
+        """Set or clear the runtime API-key bundle without mutating environment."""
+
+        if raw_keys is None:
+            saas_web.CREDENTIAL_REGISTRY.delete_credential(self._credential_name)
+        else:
+            saas_web.CREDENTIAL_REGISTRY.set_credential(
+                self._credential_name,
+                raw_keys,
+            )
 
     def _post_shrink(self, headers=None):
         """POST a minimal /shrink request and return the response."""
@@ -610,10 +642,9 @@ def _post_shrink(self, headers=None):
             headers=headers or {},
         )
 
-    def test_no_env_var_leaves_endpoints_open(self):
-        with patch.dict(os.environ):
-            os.environ.pop("CODEC_CARVER_API_KEYS", None)
-            response = self._post_shrink()
+    def test_no_registry_value_leaves_endpoints_open(self):
+        self._configure_keys(None)
+        response = self._post_shrink()
 
         self.assertEqual(response.status_code, 200)
         self.assertEqual(
@@ -622,24 +653,24 @@ def test_no_env_var_leaves_endpoints_open(self):
         )
 
     def test_missing_header_rejected_when_keys_configured(self):
-        with patch.dict(os.environ, {"CODEC_CARVER_API_KEYS": "secret-key"}):
-            response = self._post_shrink()
+        self._configure_keys("secret-key")
+        response = self._post_shrink()
 
         self.assertEqual(response.status_code, 401)
         self.assertEqual(response.json(), {"error": "Invalid or missing API key"})
         self.assertNotIn("secret-key", response.text)
 
     def test_wrong_key_rejected(self):
-        with patch.dict(os.environ, {"CODEC_CARVER_API_KEYS": "secret-key"}):
-            response = self._post_shrink(headers={"X-API-Key": "wrong-key"})
+        self._configure_keys("secret-key")
+        response = self._post_shrink(headers={"X-API-Key": "wrong-key"})
 
         self.assertEqual(response.status_code, 401)
         self.assertEqual(response.json(), {"error": "Invalid or missing API key"})
         self.assertNotIn("secret-key", response.text)
 
     def test_correct_key_reaches_handler(self):
-        with patch.dict(os.environ, {"CODEC_CARVER_API_KEYS": "secret-key"}):
-            response = self._post_shrink(headers={"X-API-Key": "secret-key"})
+        self._configure_keys("secret-key")
+        response = self._post_shrink(headers={"X-API-Key": "secret-key"})
 
         self.assertEqual(response.status_code, 200)
         self.assertEqual(
@@ -648,51 +679,51 @@ def test_correct_key_reaches_handler(self):
         )
 
     def test_get_ui_always_open_without_key(self):
-        with patch.dict(os.environ, {"CODEC_CARVER_API_KEYS": "secret-key"}):
-            response = client.get("/")
+        self._configure_keys("secret-key")
+        response = client.get("/")
 
         self.assertEqual(response.status_code, 200)
         self.assertIn(b"Codec Carver SaaS", response.content)
 
     def test_job_api_requires_key_when_configured(self):
-        with patch.dict(os.environ, {"CODEC_CARVER_API_KEYS": "secret-key"}):
-            response = client.get("/jobs/missing")
-            allowed = client.get("/jobs/missing", headers={"X-API-Key": "secret-key"})
+        self._configure_keys("secret-key")
+        response = client.get("/jobs/missing")
+        allowed = client.get("/jobs/missing", headers={"X-API-Key": "secret-key"})
 
         self.assertEqual(response.status_code, 401)
         self.assertEqual(response.json(), {"error": "Invalid or missing API key"})
         self.assertEqual(allowed.status_code, 404)
 
     def test_multiple_comma_separated_keys_all_valid(self):
-        with patch.dict(os.environ, {"CODEC_CARVER_API_KEYS": "key-one,key-two,key-three"}):
-            for key in ("key-one", "key-two", "key-three"):
-                response = self._post_shrink(headers={"X-API-Key": key})
-                self.assertEqual(response.status_code, 200, key)
-            rejected = self._post_shrink(headers={"X-API-Key": "key-four"})
+        self._configure_keys("key-one,key-two,key-three")
+        for key in ("key-one", "key-two", "key-three"):
+            response = self._post_shrink(headers={"X-API-Key": key})
+            self.assertEqual(response.status_code, 200, key)
+        rejected = self._post_shrink(headers={"X-API-Key": "key-four"})
 
         self.assertEqual(rejected.status_code, 401)
 
     def test_whitespace_around_keys_is_stripped(self):
-        with patch.dict(os.environ, {"CODEC_CARVER_API_KEYS": "  key-one , key-two  "}):
-            response = self._post_shrink(headers={"X-API-Key": "key-one"})
-            self.assertEqual(response.status_code, 200)
-            response = self._post_shrink(headers={"X-API-Key": "key-two"})
-            self.assertEqual(response.status_code, 200)
-            rejected = self._post_shrink(headers={"X-API-Key": " key-one "})
+        self._configure_keys("  key-one , key-two  ")
+        response = self._post_shrink(headers={"X-API-Key": "key-one"})
+        self.assertEqual(response.status_code, 200)
+        response = self._post_shrink(headers={"X-API-Key": "key-two"})
+        self.assertEqual(response.status_code, 200)
+        rejected = self._post_shrink(headers={"X-API-Key": " key-one "})
 
         self.assertEqual(rejected.status_code, 401)
 
     def test_empty_entries_are_ignored(self):
-        with patch.dict(os.environ, {"CODEC_CARVER_API_KEYS": "key-one,,  ,"}):
-            response = self._post_shrink(headers={"X-API-Key": "key-one"})
-            self.assertEqual(response.status_code, 200)
-            rejected = self._post_shrink(headers={"X-API-Key": ""})
+        self._configure_keys("key-one,,  ,")
+        response = self._post_shrink(headers={"X-API-Key": "key-one"})
+        self.assertEqual(response.status_code, 200)
+        rejected = self._post_shrink(headers={"X-API-Key": ""})
 
         self.assertEqual(rejected.status_code, 401)
 
     def test_only_empty_entries_leave_endpoints_open(self):
-        with patch.dict(os.environ, {"CODEC_CARVER_API_KEYS": " , ,"}):
-            response = self._post_shrink()
+        self._configure_keys(" , ,")
+        response = self._post_shrink()
 
         self.assertEqual(response.status_code, 200)
         self.assertEqual(
@@ -701,11 +732,19 @@ def test_only_empty_entries_leave_endpoints_open(self):
         )
 
     def test_get_configured_api_keys_parsing(self):
-        with patch.dict(os.environ, {"CODEC_CARVER_API_KEYS": " a ,, b ,"}):
-            self.assertEqual(saas_web.get_configured_api_keys(), ["a", "b"])
-        with patch.dict(os.environ):
-            os.environ.pop("CODEC_CARVER_API_KEYS", None)
-            self.assertEqual(saas_web.get_configured_api_keys(), [])
+        self._configure_keys(" a ,, b ,")
+        self.assertEqual(saas_web.get_configured_api_keys(), ["a", "b"])
+        self._configure_keys(None)
+        self.assertEqual(saas_web.get_configured_api_keys(), [])
+
+    def test_environment_mutation_is_not_a_runtime_credential_source(self):
+        self._configure_keys("registry-key")
+        with patch.dict(os.environ, {"CODEC_CARVER_API_KEYS": "environment-key"}):
+            rejected = self._post_shrink(headers={"X-API-Key": "environment-key"})
+            allowed = self._post_shrink(headers={"X-API-Key": "registry-key"})
+
+        self.assertEqual(rejected.status_code, 401)
+        self.assertEqual(allowed.status_code, 200)
 
     def test_wrong_key_with_non_ascii_characters_handled_safely(self):
         import asyncio
@@ -722,8 +761,8 @@ async def call_next(req):
             "headers": [(b"x-api-key", "secret-key🚀".encode("utf-8"))]
         })
 
-        with patch.dict(os.environ, {"CODEC_CARVER_API_KEYS": "secret-key"}):
-            response = asyncio.run(saas_web.require_api_key(mock_request, call_next))
+        self._configure_keys("secret-key")
+        response = asyncio.run(saas_web.require_api_key(mock_request, call_next))
 
         self.assertEqual(response.status_code, 401)
         import json

From 0794014006acf2c433043f001508cc6b12abc668 Mon Sep 17 00:00:00 2001
From: Seongho Bae <me@seonghobae.me>
Date: Fri, 7 Aug 2026 08:19:52 +0900
Subject: [PATCH 07/20] docs(security): doctor API-key registry lifecycle and
 evidence

---
 docs/doctoring/api-key-credential-registry.md | 93 +++++++++++++++++++
 1 file changed, 93 insertions(+)
 create mode 100644 docs/doctoring/api-key-credential-registry.md

diff --git a/docs/doctoring/api-key-credential-registry.md b/docs/doctoring/api-key-credential-registry.md
new file mode 100644
index 00000000..54ef43b4
--- /dev/null
+++ b/docs/doctoring/api-key-credential-registry.md
@@ -0,0 +1,93 @@
+# API-key credential registry boundary
+
+## Decision
+
+Codec Carver treats `CODEC_CARVER_API_KEYS` as a deployment-time bootstrap
+transport, not as the runtime credential source. During service initialization,
+`bootstrap_codec_carver_api_keys()` copies the value into the process-local
+`CredentialRegistry`. Request middleware subsequently reads only the registry.
+Changing the process environment after bootstrap therefore cannot silently
+change the authentication boundary.
+
+This design intentionally does **not** persist API keys in the SQLite job store.
+The job database has no credential-storage contract or at-rest encryption key,
+so writing reusable API-key material there would create a new plaintext-at-rest
+secret surface. The registry interface is deliberately small so a future
+external secret manager, sidecar, or platform KV adapter can populate the same
+runtime boundary without changing request authentication logic.
+
+## Security properties
+
+- Credential material is not read from mutable process environment variables on
+  each request.
+- Registry access is protected by a re-entrant lock so concurrent requests and
+  controlled credential replacement cannot observe partially updated state.
+- Credential names are validated and blank names fail closed.
+- API-key comparison remains constant-time through `hmac.compare_digest` over
+  UTF-8 bytes, so non-ASCII attacker input is rejected without the `TypeError`
+  denial-of-service path that motivated PR #361.
+- An absent or explicitly empty API-key bundle preserves the existing opt-in
+  authentication mode. Deployments that require mandatory authentication must
+  provision a non-empty bundle before service start and verify the protected
+  endpoint contract during readiness checks.
+- Runtime rotation is an explicit control-plane action. Operators must update
+  the registry through the approved bootstrap/secret-manager integration or
+  restart the service with newly provisioned bootstrap material; editing the
+  process environment behind a running worker is not a supported rotation path.
+
+## Verification contract
+
+`tests/test_credential_registry.py` pins registry set/get/delete behavior,
+rejects ambiguous names, and proves that mutating the bootstrap mapping after a
+copy does not alter runtime state. `TestApiKeyAuth` in
+`tests/test_saas_web.py` uses registry fixtures rather than environment-variable
+patches and proves that an environment-only replacement key is rejected while
+the registered key continues to authenticate. The pre-existing non-ASCII test
+continues to prove that malformed Unicode input returns HTTP 401 instead of
+raising through `compare_digest`.
+
+The first registry regression test was committed before the implementation so
+review history preserves the RED→GREEN sequence. The final merge gate remains
+exact-head CI/security checks plus repository-required independent approval.
+
+## Standards and research basis
+
+NIST SP 800-57 Part 1 Revision 5 remains the current final NIST general
+key-management recommendation as of August 7, 2026. NIST has published Revision
+6 only as an Initial Public Draft; it adds an expanded discussion of keying
+material storage, but it is not substituted for the current final publication.
+The separation between bootstrap transport and runtime credential use follows
+the broader lifecycle principle that sensitive keying material requires
+explicit protection, use, replacement, and destruction boundaries.
+
+OWASP ASVS 5.0 secret-management requirements call for a secrets-management
+solution to create, store, control access to, and destroy backend secrets. The
+process-local registry in this PR is not represented as a full enterprise vault;
+it is the application boundary that removes per-request environment reads and
+provides a stable integration point for such a vault or platform KV.
+
+Krause et al. (2023) found that secret leakage is a recurring developer problem
+and emphasized low-adoption-cost prevention and remediation mechanisms. Keeping
+bootstrap compatibility while moving request-time consumption behind one small
+registry minimizes deployment disruption while closing the evidenced runtime
+configuration coupling.
+
+## References
+
+Barker, E. (2020). *Recommendation for key management: Part 1—General*
+(NIST Special Publication 800-57 Part 1 Revision 5). National Institute of
+Standards and Technology. https://doi.org/10.6028/NIST.SP.800-57pt1r5
+
+Barker, E., & Barker, W. (2025). *Recommendation for key management: Part 1—General*
+(Initial Public Draft, NIST Special Publication 800-57 Part 1 Revision 6).
+National Institute of Standards and Technology.
+https://doi.org/10.6028/NIST.SP.800-57pt1r6.ipd
+
+Krause, A., Klemmer, J. H., Huaman, N., Wermke, D., Acar, Y., & Fahl, S.
+(2023). Pushed by accident: A mixed-methods study on strategies of handling
+secret information in source code repositories. In *32nd USENIX Security
+Symposium (USENIX Security 23)* (pp. 2527–2544). USENIX Association.
+https://www.usenix.org/conference/usenixsecurity23/presentation/krause
+
+OWASP Foundation. (2025). *OWASP Application Security Verification Standard
+5.0: V13.3 Secret management*. https://cornucopia.owasp.org/taxonomy/asvs-5.0/13-configuration/03-secret-management

From 93f92c93e414b290ca067af299d34e587c6f083b Mon Sep 17 00:00:00 2001
From: Seongho Bae <me@seonghobae.me>
Date: Fri, 7 Aug 2026 08:22:01 +0900
Subject: [PATCH 08/20] test(storage): pin owned result root and independent
 retention

---
 tests/test_result_retention.py | 191 +++++++++++++++++++++++++++++++++
 1 file changed, 191 insertions(+)
 create mode 100644 tests/test_result_retention.py

diff --git a/tests/test_result_retention.py b/tests/test_result_retention.py
new file mode 100644
index 00000000..f354aec2
--- /dev/null
+++ b/tests/test_result_retention.py
@@ -0,0 +1,191 @@
+"""Regression tests for owned result storage and download-independent cleanup."""
+
+from __future__ import annotations
+
+import io
+import tempfile
+import unittest
+from datetime import datetime, timedelta, timezone
+from pathlib import Path
+from unittest.mock import MagicMock, patch
+
+try:
+    from fastapi.testclient import TestClient
+
+    import saas_web
+    from job_store import JobStore
+    from media_shrinker import ConversionResult
+
+    _HAS_FASTAPI = True
+except ImportError:
+    _HAS_FASTAPI = False
+
+
+@unittest.skipUnless(_HAS_FASTAPI, "fastapi not installed (optional integration dependency)")
+class ResultRetentionTests(unittest.TestCase):
+    """Pin owned-root creation and bounded retention for completed job outputs."""
+
+    def setUp(self) -> None:
+        """Create isolated job metadata and result roots for every test."""
+
+        self._temporary_directory = tempfile.TemporaryDirectory()
+        self.addCleanup(self._temporary_directory.cleanup)
+        self.test_root = Path(self._temporary_directory.name)
+        self.store = JobStore(str(self.test_root / "jobs.sqlite3"))
+        self.previous_results_root = saas_web._RESULTS_ROOT
+        self.addCleanup(setattr, saas_web, "_RESULTS_ROOT", self.previous_results_root)
+        saas_web._RESULTS_ROOT = self.test_root / "owned_results"
+        saas_web._RESULTS_ROOT.mkdir()
+
+    def _create_done_job(
+        self,
+        job_id: str,
+        *,
+        updated_at: datetime,
+        output_path: Path | None,
+    ) -> None:
+        """Insert one completed job with deterministic timestamps and output metadata."""
+
+        created_at = updated_at - timedelta(minutes=1)
+        self.store.create(job_id, temp_dir="", now=created_at)
+        self.store.set_status(
+            job_id,
+            "done",
+            now=updated_at,
+            output_path=str(output_path) if output_path is not None else None,
+            output_name=output_path.name if output_path is not None else None,
+        )
+
+    def test_results_root_uses_mkdtemp_once_and_reuses_owned_root(self) -> None:
+        """The service creates one unpredictable owned root instead of a fixed /tmp name."""
+
+        saas_web._RESULTS_ROOT = None
+        generated_root = self.test_root / "generated_owner_root"
+        generated_root.mkdir()
+        with patch("saas_web.tempfile.mkdtemp", return_value=str(generated_root)) as mock_mkdtemp:
+            first = saas_web._get_results_root()
+            second = saas_web._get_results_root()
+
+        self.assertEqual(first, generated_root.resolve())
+        self.assertIs(first, second)
+        mock_mkdtemp.assert_called_once_with(prefix="codec_carver_results_")
+
+    def test_expired_cleanup_removes_old_file_and_job_but_preserves_fresh_result(self) -> None:
+        """Retention cleanup removes expired bytes and metadata independently of download."""
+
+        now = datetime(2026, 8, 7, 0, 0, tzinfo=timezone.utc)
+        expired_output = saas_web._get_results_root() / "expired.flac"
+        fresh_output = saas_web._get_results_root() / "fresh.flac"
+        expired_output.write_bytes(b"expired")
+        fresh_output.write_bytes(b"fresh")
+        self._create_done_job(
+            "expired-job",
+            updated_at=now - timedelta(seconds=saas_web.RESULT_RETENTION_SECONDS + 1),
+            output_path=expired_output,
+        )
+        self._create_done_job(
+            "fresh-job",
+            updated_at=now - timedelta(seconds=saas_web.RESULT_RETENTION_SECONDS - 1),
+            output_path=fresh_output,
+        )
+
+        removed = saas_web._cleanup_expired_results(self.store, now=now)
+
+        self.assertEqual(removed, 1)
+        self.assertFalse(expired_output.exists())
+        self.assertIsNone(self.store.get("expired-job"))
+        self.assertTrue(fresh_output.exists())
+        self.assertIsNotNone(self.store.get("fresh-job"))
+
+    def test_expired_cleanup_never_deletes_file_outside_owned_root(self) -> None:
+        """A compromised metadata path cannot make retention unlink an external file."""
+
+        now = datetime(2026, 8, 7, 0, 0, tzinfo=timezone.utc)
+        outside_output = self.test_root / "outside-secret.flac"
+        outside_output.write_bytes(b"must-survive")
+        self._create_done_job(
+            "outside-job",
+            updated_at=now - timedelta(seconds=saas_web.RESULT_RETENTION_SECONDS + 1),
+            output_path=outside_output,
+        )
+
+        removed = saas_web._cleanup_expired_results(self.store, now=now)
+
+        self.assertEqual(removed, 1)
+        self.assertTrue(outside_output.exists())
+        self.assertIsNone(self.store.get("outside-job"))
+
+    def test_expired_cleanup_removes_metadata_without_output_path(self) -> None:
+        """Expired done records without result bytes are still retired from metadata."""
+
+        now = datetime(2026, 8, 7, 0, 0, tzinfo=timezone.utc)
+        self._create_done_job(
+            "metadata-only",
+            updated_at=now - timedelta(seconds=saas_web.RESULT_RETENTION_SECONDS + 1),
+            output_path=None,
+        )
+
+        removed = saas_web._cleanup_expired_results(self.store, now=now)
+
+        self.assertEqual(removed, 1)
+        self.assertIsNone(self.store.get("metadata-only"))
+
+    def test_malformed_timestamp_is_preserved_for_operator_diagnosis(self) -> None:
+        """Malformed durable metadata is logged and retained instead of guessed expired."""
+
+        now = datetime(2026, 8, 7, 0, 0, tzinfo=timezone.utc)
+        self.store.create("malformed-job", temp_dir="", now=now)
+        self.store.set_status("malformed-job", "done", now=now)
+        with self.store._lock, self.store._connect() as connection:
+            connection.execute(
+                "UPDATE jobs SET updated_at = ? WHERE id = ?",
+                ("not-a-timestamp", "malformed-job"),
+            )
+
+        with self.assertLogs("saas_web", level="WARNING"):
+            removed = saas_web._cleanup_expired_results(self.store, now=now)
+
+        self.assertEqual(removed, 0)
+        self.assertIsNotNone(self.store.get("malformed-job"))
+
+    @patch("saas_web.media_shrinker.convert_file")
+    @patch("saas_web._get_results_root", side_effect=OSError("no safe result root"))
+    def test_batch_result_persistence_failure_cleans_workspace(
+        self,
+        _mock_results_root,
+        mock_convert_file,
+    ) -> None:
+        """Batch persistence failure returns 500 and removes the request workspace."""
+
+        created_workspaces: list[Path] = []
+        real_mkdtemp = tempfile.mkdtemp
+
+        def tracked_mkdtemp(*args, **kwargs):
+            path = Path(real_mkdtemp(*args, dir=self.test_root, **kwargs))
+            created_workspaces.append(path)
+            return str(path)
+
+        def fake_convert(*, source, root, output_dir, target_bytes):
+            output = Path(output_dir) / "out.flac"
+            output.write_bytes(b"converted")
+            result = MagicMock(spec=ConversionResult)
+            result.output_path = output
+            return [result]
+
+        mock_convert_file.side_effect = fake_convert
+        with patch("saas_web.tempfile.mkdtemp", side_effect=tracked_mkdtemp):
+            client = TestClient(saas_web.app)
+            response = client.post(
+                "/shrink-batch",
+                files=[("files", ("a.wav", io.BytesIO(b"audio"), "audio/wav"))],
+                data={"target_bytes": 10000},
+            )
+
+        self.assertEqual(response.status_code, 500)
+        self.assertEqual(response.json(), {"error": "Upload processing failed"})
+        self.assertTrue(created_workspaces)
+        self.assertTrue(all(not path.exists() for path in created_workspaces))
+
+
+if __name__ == "__main__":  # pragma: no cover
+    unittest.main()

From 6a1025705b6cd9e31f28afb90f06b35089f17930 Mon Sep 17 00:00:00 2001
From: Seongho Bae <me@seonghobae.me>
Date: Fri, 7 Aug 2026 08:24:05 +0900
Subject: [PATCH 09/20] fix(storage): own result root and retire expired job
 outputs

---
 saas_web.py | 144 +++++++++++++++++++++++++++++++++++++++++++---------
 1 file changed, 121 insertions(+), 23 deletions(-)

diff --git a/saas_web.py b/saas_web.py
index 5b1d3e02..5c701ed7 100644
--- a/saas_web.py
+++ b/saas_web.py
@@ -6,9 +6,10 @@
 import os
 import shutil
 import tempfile
+import threading
 import uuid
 import zipfile
-from datetime import datetime, timezone
+from datetime import datetime, timedelta, timezone
 from pathlib import Path
 from fastapi import FastAPI, UploadFile, File, BackgroundTasks, Form, Request
 from fastapi.responses import HTMLResponse, FileResponse, JSONResponse
@@ -33,6 +34,12 @@
 # A shrink target larger than the biggest accepted upload is meaningless; cap it
 # to keep numeric input bounded.
 MAX_TARGET_BYTES = MAX_UPLOAD_BYTES
+# Completed async results are retired on subsequent job work even when a client
+# never downloads them.  This bounds attacker-driven accumulation without a
+# per-request background scheduler or a second persistence dependency.
+RESULT_RETENTION_SECONDS = 24 * 60 * 60
+_RESULTS_ROOT: Path | None = None
+_RESULTS_ROOT_LOCK = threading.Lock()
 # This service only processes audio/video. Uploaded files are never executed or
 # served as web content — they are handed to ffmpeg, which rejects non-media —
 # but validating the declared content type rejects obviously-wrong uploads early.
@@ -427,6 +434,69 @@ def cleanup_temp_dir(temp_dir_path: Path):
         shutil.rmtree(temp_dir_path, ignore_errors=True)
 
 
+def _get_results_root() -> Path:
+    """Return the process-owned unpredictable root used for persisted results.
+
+    ``tempfile.mkdtemp`` atomically creates the directory with an unpredictable
+    name and user-restricted permissions. The first successful call owns the
+    root for the life of this process; later calls reuse the exact same path so
+    batch persistence, async job lookup, retention, and cleanup share one trust
+    boundary.
+    """
+
+    global _RESULTS_ROOT
+    with _RESULTS_ROOT_LOCK:
+        if _RESULTS_ROOT is None:
+            _RESULTS_ROOT = Path(
+                tempfile.mkdtemp(prefix="codec_carver_results_")
+            ).resolve()
+        return _RESULTS_ROOT
+
+
+def _cleanup_expired_results(store: JobStore, *, now: datetime) -> int:
+    """Retire expired completed jobs and their owned result files.
+
+    Cleanup is opportunistic but download-independent: callers invoke it before
+    accepting new async work and after completing a job. Consequently repeated
+    submissions cannot accumulate completed results indefinitely merely by
+    avoiding the download endpoint. Files referenced outside the process-owned
+    result root are never unlinked even if durable metadata is compromised; the
+    expired metadata record is still retired.
+
+    Args:
+        store: Durable job store whose completed records are eligible.
+        now: Aware timestamp used to compute the deterministic retention cutoff.
+
+    Returns:
+        Number of expired job records removed.
+    """
+
+    cutoff = now - timedelta(seconds=RESULT_RETENTION_SECONDS)
+    results_root = _get_results_root()
+    removed = 0
+    for job in store.list_jobs(status="done"):
+        try:
+            updated_at = datetime.fromisoformat(job["updated_at"])
+        except (TypeError, ValueError):
+            logger.warning("Skipping malformed completed-job timestamp for %s", job["id"])
+            continue
+        try:
+            is_expired = updated_at <= cutoff
+        except TypeError:
+            logger.warning("Skipping timezone-incompatible completed-job timestamp for %s", job["id"])
+            continue
+        if not is_expired:
+            continue
+        output_path_text = job.get("output_path")
+        if output_path_text:
+            output_path = Path(output_path_text).resolve()
+            if output_path.is_relative_to(results_root):
+                output_path.unlink(missing_ok=True)
+        store.delete(job["id"])
+        removed += 1
+    return removed
+
+
 def _zip_outputs(outputs: list[Path], dest_dir: Path, archive_name: str) -> Path:
     """Bundle multiple generated outputs into a single (uncompressed) zip archive."""
     archive_path = dest_dir / archive_name
@@ -678,21 +748,20 @@ def shrink_media_batch(
                 "results.json",
                 json.dumps({"target_bytes": target_bytes, "results": manifest}, indent=2),
             )
+
+        results_root = _get_results_root()
+        persistent_path = results_root / f"batch_{uuid.uuid4().hex}.zip"
+        shutil.move(str(zip_path), str(persistent_path))
     except Exception:
         cleanup_temp_dir(temp_dir_path)
-        logger.exception("Failed to build batch archive")
+        logger.exception("Failed to build or persist batch archive")
         return JSONResponse(status_code=500, content={"error": "Upload processing failed"})
 
-    import shutil
-    import uuid
-    import starlette.background
-    results_dir = Path(tempfile.gettempdir()) / "codec_carver_results"
-    results_dir.mkdir(parents=True, exist_ok=True)
-    persistent_path = results_dir / f"batch_{uuid.uuid4().hex}.zip"
-    shutil.move(str(zip_path), str(persistent_path))
     cleanup_temp_dir(temp_dir_path)
 
     def cleanup_batch_result():
+        """Delete the persisted batch archive after its response completes."""
+
         if persistent_path.exists():
             persistent_path.unlink(missing_ok=True)
 
@@ -700,7 +769,7 @@ def cleanup_batch_result():
         path=persistent_path,
         filename="codec_carver_batch.zip",
         media_type="application/zip",
-        background=starlette.background.BackgroundTask(cleanup_batch_result)
+        background=__import__("starlette.background").background.BackgroundTask(cleanup_batch_result)
     )
 
 # --- Async job model --------------------------------------------------------
@@ -774,21 +843,32 @@ def _run_job(
             outputs, temp_dir_path, source_path.stem + "_shrunk.zip"
         )
         try:
-            import shutil
-            results_dir = Path(tempfile.gettempdir()) / "codec_carver_results"
-            results_dir.mkdir(parents=True, exist_ok=True)
-            persistent_path = results_dir / f"{job_id}_{output_path.name}"
+            results_root = _get_results_root()
+            persistent_path = results_root / f"{job_id}_{output_path.name}"
             shutil.move(str(output_path), str(persistent_path))
 
+            completed_at = _now()
             store.set_status(
                 job_id,
                 "done",
-                now=_now(),
+                now=completed_at,
                 output_path=str(persistent_path),
                 output_name=output_path.name,
             )
+            _cleanup_expired_results(store, now=completed_at)
         except KeyError:
             logger.error("Job %s disappeared while recording result", job_id)
+        except Exception:
+            logger.exception("Failed to persist or retain job result for %s", job_id)
+            try:
+                store.set_status(
+                    job_id,
+                    "failed",
+                    now=_now(),
+                    error="Result storage failed",
+                )
+            except KeyError:
+                logger.error("Job %s disappeared while recording result-storage failure", job_id)
         finally:
             cleanup_temp_dir(temp_dir_path)
     else:
@@ -816,6 +896,16 @@ def submit_job(
     if error is not None:
         return JSONResponse(status_code=400, content={"error": error})
 
+    store = _get_job_store()
+    try:
+        _cleanup_expired_results(store, now=_now())
+    except Exception:
+        logger.exception("Result retention cleanup failed before accepting new work")
+        return JSONResponse(
+            status_code=503,
+            content={"error": "Result storage unavailable"},
+        )
+
     try:
         temp_dir_path, input_dir, output_dir, source_path = _persist_upload(file)
     except Exception:
@@ -826,7 +916,7 @@ def submit_job(
 
     job_id = uuid.uuid4().hex
     try:
-        _get_job_store().create(job_id, temp_dir=str(temp_dir_path), now=_now())
+        store.create(job_id, temp_dir=str(temp_dir_path), now=_now())
     except ValueError:
         cleanup_temp_dir(temp_dir_path)
         logger.exception("Failed to create async job record")
@@ -850,7 +940,7 @@ def job_status(job_id: str):
 
 
 def _cleanup_job(job_id: str) -> None:
-    """Forget a job and remove its temporary workspace."""
+    """Forget a job and remove its temporary workspace and owned output."""
     store = _get_job_store()
     job = store.get(job_id)
     store.delete(job_id)
@@ -859,15 +949,23 @@ def _cleanup_job(job_id: str) -> None:
             cleanup_temp_dir(Path(job["temp_dir"]))
         output_path_text = job.get("output_path")
         if output_path_text:
-            output_path = Path(output_path_text)
-            if output_path.exists():
+            output_path = Path(output_path_text).resolve()
+            results_root = _get_results_root()
+            if output_path.is_relative_to(results_root):
                 output_path.unlink(missing_ok=True)
 
 
 @app.get("/jobs/{job_id}/result")
 def job_result(job_id: str, background_tasks: BackgroundTasks):
     """Download a finished job's output, then clean up its workspace."""
-    job = _get_job_store().get(job_id)
+    store = _get_job_store()
+    try:
+        _cleanup_expired_results(store, now=_now())
+    except Exception:
+        logger.exception("Result retention cleanup failed before result lookup")
+        return JSONResponse(status_code=503, content={"error": "Result storage unavailable"})
+
+    job = store.get(job_id)
     if job is None:
         return JSONResponse(status_code=404, content={"error": "Unknown job"})
     if job["status"] != "done":
@@ -875,7 +973,7 @@ def job_result(job_id: str, background_tasks: BackgroundTasks):
             status_code=409, content={"error": f"Job is {job['status']}"}
         )
     # Defense in depth: only ever serve a regular file that lives inside this
-    # job's own temp workspace. `job_id` is an opaque store key and is never
+    # process's owned result root. `job_id` is an opaque store key and is never
     # used to build a path, but confining the served path makes traversal
     # impossible even if the store were ever populated from untrusted data.
     output_path_text = job.get("output_path")
@@ -884,9 +982,9 @@ def job_result(job_id: str, background_tasks: BackgroundTasks):
             status_code=410, content={"error": "Result no longer available"}
         )
     output_path = Path(output_path_text).resolve()
-    results_dir = (Path(tempfile.gettempdir()) / "codec_carver_results").resolve()
+    results_root = _get_results_root()
 
-    if not output_path.is_relative_to(results_dir) or not output_path.is_file():
+    if not output_path.is_relative_to(results_root) or not output_path.is_file():
         return JSONResponse(
             status_code=410, content={"error": "Result no longer available"}
         )

From 9d7c9edd8a317a0ff59719e464af336c047f6c2d Mon Sep 17 00:00:00 2001
From: Seongho Bae <me@seonghobae.me>
Date: Fri, 7 Aug 2026 08:25:00 +0900
Subject: [PATCH 10/20] docs(security): record owned result retention limits

---
 .jules/sentinel.md | 5 +++++
 1 file changed, 5 insertions(+)

diff --git a/.jules/sentinel.md b/.jules/sentinel.md
index 84f09890..d33b13f1 100644
--- a/.jules/sentinel.md
+++ b/.jules/sentinel.md
@@ -68,3 +68,8 @@
 **취약점:** `saas_web.py`의 `/jobs` 엔드포인트에서 생성된 임시 디렉토리가 작업 결과 다운로드 시에만 삭제되도록 구현되어 있어, 공격자가 무수히 많은 작업을 요청하고 다운로드하지 않을 경우 디스크 공간이 고갈되는 서비스 거부(DoS) 취약점이 있었습니다.
 **학습:** 백그라운드 작업에서 생성되는 리소스는 클라이언트의 후속 액션(다운로드 등)에만 의존해서는 안 되며, 작업 완료 시 즉시 정리할 수 있는 구조를 갖춰야 디스크 고갈 공격을 방지할 수 있음을 배웠습니다.
 **예방:** 작업 완료 시 즉각적으로 결과물을 영구 저장소(`codec_carver_results`)로 이동한 후 무거운 임시 디렉토리는 바로 정리하고, 클라이언트가 결과를 가져간 후 영구 저장소의 파일만 삭제하도록 생명 주기를 분리하여 관리해야 합니다.
+
+## 2026-08-07 - [Sentinel: Owned result storage and independent retention]
+**Vulnerability:** 완료된 비동기 결과를 예측 가능한 공유 임시 디렉터리에 저장하고 다운로드에만 삭제를 의존하면, 선점된 디렉터리/경로 신뢰 문제와 다운로드하지 않는 반복 제출에 의한 디스크 고갈(CWE-377, CWE-400, CWE-770)이 발생할 수 있습니다.
+**Learning:** 결과 저장소는 `tempfile.mkdtemp()`가 생성한 프로세스 소유의 예측 불가능한 루트를 단일 신뢰 경계로 재사용해야 합니다. 다운로드 후 삭제만으로는 보존 정책이 아니므로, 완료 시각을 기준으로 한 별도 TTL 정리가 신규 작업 수락 전과 작업 완료 후 실행되어야 합니다. 보존 정리는 저장된 경로가 소유 결과 루트 내부일 때만 파일을 삭제하며, 외부 경로를 가리키는 손상된 메타데이터는 파일을 건드리지 않고 만료 레코드만 제거합니다.
+**Prevention:** 완료 결과의 기본 보존 기간은 24시간(`RESULT_RETENTION_SECONDS`)이며, 새 `/jobs` 요청을 수락하기 전에 만료 결과 정리가 성공해야 합니다. 각 작업 완료 후에도 같은 정리를 수행하므로 공격자가 결과를 다운로드하지 않아도 오래된 파일과 해당 `JobStore` 레코드는 함께 제거됩니다. 서비스가 완전히 유휴한 동안에는 TTL을 넘긴 파일이 다음 작업/결과 접근까지 남을 수 있지만 추가 디스크 소비도 발생하지 않습니다. 동기 배치 결과는 동일한 소유 루트에 이동한 뒤 응답 완료 시 즉시 삭제하며, 결과 루트 생성 또는 이동 실패 시 요청 작업공간을 정리하고 표준 오류 응답으로 실패합니다. 다운로드 후 `_cleanup_job` 삭제는 TTL과 별개의 보조 정리로 유지됩니다.
\ No newline at end of file

From e564664d6caf355e09d0a1e221a4a825ffc251a7 Mon Sep 17 00:00:00 2001
From: Seongho Bae <me@seonghobae.me>
Date: Fri, 7 Aug 2026 08:26:03 +0900
Subject: [PATCH 11/20] test(storage): expose app import failures and isolate
 auth state

---
 tests/test_result_retention.py | 45 +++++++++++++++++++++++++++++-----
 1 file changed, 39 insertions(+), 6 deletions(-)

diff --git a/tests/test_result_retention.py b/tests/test_result_retention.py
index f354aec2..1da72d10 100644
--- a/tests/test_result_retention.py
+++ b/tests/test_result_retention.py
@@ -11,22 +11,21 @@
 
 try:
     from fastapi.testclient import TestClient
-
+except ImportError:
+    _HAS_FASTAPI = False
+else:
+    _HAS_FASTAPI = True
     import saas_web
     from job_store import JobStore
     from media_shrinker import ConversionResult
 
-    _HAS_FASTAPI = True
-except ImportError:
-    _HAS_FASTAPI = False
-
 
 @unittest.skipUnless(_HAS_FASTAPI, "fastapi not installed (optional integration dependency)")
 class ResultRetentionTests(unittest.TestCase):
     """Pin owned-root creation and bounded retention for completed job outputs."""
 
     def setUp(self) -> None:
-        """Create isolated job metadata and result roots for every test."""
+        """Create isolated job metadata, credentials, and result roots for every test."""
 
         self._temporary_directory = tempfile.TemporaryDirectory()
         self.addCleanup(self._temporary_directory.cleanup)
@@ -37,6 +36,23 @@ def setUp(self) -> None:
         saas_web._RESULTS_ROOT = self.test_root / "owned_results"
         saas_web._RESULTS_ROOT.mkdir()
 
+        credential_name = saas_web.CODEC_CARVER_API_KEYS_NAME
+        previous_credential = saas_web.CREDENTIAL_REGISTRY.get_credential(credential_name)
+        saas_web.CREDENTIAL_REGISTRY.delete_credential(credential_name)
+
+        def restore_credential() -> None:
+            """Restore deployment credential state after a retention test."""
+
+            if previous_credential is None:
+                saas_web.CREDENTIAL_REGISTRY.delete_credential(credential_name)
+            else:
+                saas_web.CREDENTIAL_REGISTRY.set_credential(
+                    credential_name,
+                    previous_credential,
+                )
+
+        self.addCleanup(restore_credential)
+
     def _create_done_job(
         self,
         job_id: str,
@@ -148,6 +164,19 @@ def test_malformed_timestamp_is_preserved_for_operator_diagnosis(self) -> None:
         self.assertEqual(removed, 0)
         self.assertIsNotNone(self.store.get("malformed-job"))
 
+    def test_naive_timestamp_is_preserved_for_operator_diagnosis(self) -> None:
+        """Timezone-incompatible metadata is logged rather than guessed expired."""
+
+        now = datetime(2026, 8, 7, 0, 0, tzinfo=timezone.utc)
+        naive = datetime(2026, 8, 5, 0, 0)
+        self._create_done_job("naive-job", updated_at=naive, output_path=None)
+
+        with self.assertLogs("saas_web", level="WARNING"):
+            removed = saas_web._cleanup_expired_results(self.store, now=now)
+
+        self.assertEqual(removed, 0)
+        self.assertIsNotNone(self.store.get("naive-job"))
+
     @patch("saas_web.media_shrinker.convert_file")
     @patch("saas_web._get_results_root", side_effect=OSError("no safe result root"))
     def test_batch_result_persistence_failure_cleans_workspace(
@@ -161,11 +190,15 @@ def test_batch_result_persistence_failure_cleans_workspace(
         real_mkdtemp = tempfile.mkdtemp
 
         def tracked_mkdtemp(*args, **kwargs):
+            """Create a test-owned workspace while recording its path for cleanup proof."""
+
             path = Path(real_mkdtemp(*args, dir=self.test_root, **kwargs))
             created_workspaces.append(path)
             return str(path)
 
         def fake_convert(*, source, root, output_dir, target_bytes):
+            """Write one valid conversion result into the request output directory."""
+
             output = Path(output_dir) / "out.flac"
             output.write_bytes(b"converted")
             result = MagicMock(spec=ConversionResult)

From 058ecfc980676b67d592892575678dcebf1cbcdb Mon Sep 17 00:00:00 2001
From: Seongho Bae <me@seonghobae.me>
Date: Fri, 7 Aug 2026 08:26:28 +0900
Subject: [PATCH 12/20] docs(security): remove review-time-sensitive standards
 date

---
 docs/doctoring/api-key-credential-registry.md | 12 ++++++------
 1 file changed, 6 insertions(+), 6 deletions(-)

diff --git a/docs/doctoring/api-key-credential-registry.md b/docs/doctoring/api-key-credential-registry.md
index 54ef43b4..5e78ebde 100644
--- a/docs/doctoring/api-key-credential-registry.md
+++ b/docs/doctoring/api-key-credential-registry.md
@@ -53,12 +53,12 @@ exact-head CI/security checks plus repository-required independent approval.
 ## Standards and research basis
 
 NIST SP 800-57 Part 1 Revision 5 remains the current final NIST general
-key-management recommendation as of August 7, 2026. NIST has published Revision
-6 only as an Initial Public Draft; it adds an expanded discussion of keying
-material storage, but it is not substituted for the current final publication.
-The separation between bootstrap transport and runtime credential use follows
-the broader lifecycle principle that sensitive keying material requires
-explicit protection, use, replacement, and destruction boundaries.
+key-management recommendation. NIST has published Revision 6 only as an Initial
+Public Draft; it adds an expanded discussion of keying material storage, but it
+is not substituted for the current final publication. The separation between
+bootstrap transport and runtime credential use follows the broader lifecycle
+principle that sensitive keying material requires explicit protection, use,
+replacement, and destruction boundaries.
 
 OWASP ASVS 5.0 secret-management requirements call for a secrets-management
 solution to create, store, control access to, and destroy backend secrets. The

From 56f6c245bfe9fe806538f9428109ea713ae8785d Mon Sep 17 00:00:00 2001
From: Seongho Bae <me@seonghobae.me>
Date: Fri, 7 Aug 2026 08:28:06 +0900
Subject: [PATCH 13/20] docs(research): ground owned result storage and
 retention

---
 docs/papers/README.md | 55 ++++++++++++++++++++++++++++++++++++++++---
 1 file changed, 52 insertions(+), 3 deletions(-)

diff --git a/docs/papers/README.md b/docs/papers/README.md
index cc1d0fd7..32553033 100644
--- a/docs/papers/README.md
+++ b/docs/papers/README.md
@@ -1,10 +1,59 @@
 # Reference papers
 
+## Fuzzing
+
 - **`fuzzing-art-science-engineering-survey.pdf`** —
-  V.-T. Manès, H. Han, C. Han, S. K. Cha, M. Egele, E. J. Schwartz, M. Woo,
-  *"The Art, Science, and Engineering of Fuzzing: A Survey"*, IEEE TSE 2019
-  ([arXiv:1812.00140](https://arxiv.org/abs/1812.00140)).
+  Manès, V.-T., Han, H., Han, C., Cha, S. K., Egele, M., Schwartz, E. J., &
+  Woo, M. (2019). The art, science, and engineering of fuzzing: A survey.
+  *IEEE Transactions on Software Engineering, 47*(11), 2312–2331.
+  https://doi.org/10.1109/TSE.2019.2946563
 
   Background for the coverage-guided fuzzing approach used in `fuzz/`: the
   Atheris harnesses are libFuzzer-style greybox fuzzers as characterised in
   Section 3 of the survey.
+
+## Owned result storage and retention
+
+- Cowan, C., Beattie, S., Wright, C., & Kroah-Hartman, G. (2001). RaceGuard:
+  Kernel protection from temporary file race vulnerabilities. In *10th USENIX
+  Security Symposium (USENIX Security 01)*. USENIX Association.
+  https://www.usenix.org/conference/10th-usenix-security-symposium/raceguard-kernel-protection-temporary-file-race
+
+  This peer-reviewed systems-security paper establishes why temporary resources
+  created through non-atomic, predictable names are vulnerable to attacker races.
+  It supports Codec Carver's use of an atomically created, unpredictable
+  process-owned result root rather than a fixed shared `/tmp` directory name.
+
+- Python Software Foundation. (2026). *tempfile — Generate temporary files and
+  directories*. Python documentation.
+  https://docs.python.org/3/library/tempfile.html
+
+  The standard-library contract documents `mkdtemp()` as the secure primitive
+  for creating temporary directories and makes cleanup the caller's explicit
+  responsibility. Codec Carver therefore creates its result root with
+  `tempfile.mkdtemp()` and owns deletion through its result lifecycle instead of
+  assuming the operating system will remove application artifacts immediately.
+
+- systemd project. (2026). *Using /tmp/ and /var/tmp/ safely*.
+  https://systemd.io/TEMPORARY_DIRECTORIES/
+
+  The primary systemd guidance describes private temporary namespaces/directories
+  and age-based cleanup for temporary data. It supports treating temporary
+  storage as an explicit ownership and lifecycle boundary rather than a globally
+  trusted shared directory.
+
+- Google Cloud. (2026). *Cleanup policy overview: Artifact Registry*.
+  https://cloud.google.com/artifact-registry/docs/repositories/cleanup-policy-overview
+
+  The primary service documentation defines age-based delete policies for
+  artifacts. Codec Carver applies the same lifecycle principle locally: completed
+  async outputs receive a bounded retention window and are eligible for deletion
+  independently of whether a client ever downloads them. The local implementation
+  additionally removes the corresponding durable job record so filesystem bytes
+  and metadata do not diverge.
+
+These references justify the security and lifecycle *principles*, not the exact
+24-hour product value. `RESULT_RETENTION_SECONDS` is a Codec Carver operational
+policy chosen to bound storage consumption while leaving a practical download
+window; changing that value requires product/operations review plus retention
+regression tests.

From dcacd238f4111e48abd888b6fc459d33495f5cab Mon Sep 17 00:00:00 2001
From: Seongho Bae <me@seonghobae.me>
Date: Fri, 7 Aug 2026 08:28:31 +0900
Subject: [PATCH 14/20] test(security): cover invalid credential value branch

---
 tests/test_credential_registry.py | 6 ++++++
 1 file changed, 6 insertions(+)

diff --git a/tests/test_credential_registry.py b/tests/test_credential_registry.py
index 67630e5b..04130eb6 100644
--- a/tests/test_credential_registry.py
+++ b/tests/test_credential_registry.py
@@ -39,6 +39,12 @@ def test_blank_credential_name_is_rejected(self) -> None:
         with self.assertRaises(ValueError):
             self.registry.delete_credential("\t")
 
+    def test_non_string_credential_value_is_rejected(self) -> None:
+        """Registry values must remain immutable strings suitable for secret transport."""
+
+        with self.assertRaises(TypeError):
+            self.registry.set_credential("service_api_key", object())  # type: ignore[arg-type]
+
     def test_bootstrap_copies_api_keys_without_following_environment_mutation(self) -> None:
         """Environment transport is copied once; later mutation cannot change runtime state."""
 

From 16887cfbbde084e15c0a831c3c540cb7f9cfcc8d Mon Sep 17 00:00:00 2001
From: Seongho Bae <me@seonghobae.me>
Date: Fri, 7 Aug 2026 08:28:57 +0900
Subject: [PATCH 15/20] test(storage): pin result persistence failure isolation

---
 tests/test_result_retention_failures.py | 131 ++++++++++++++++++++++++
 1 file changed, 131 insertions(+)
 create mode 100644 tests/test_result_retention_failures.py

diff --git a/tests/test_result_retention_failures.py b/tests/test_result_retention_failures.py
new file mode 100644
index 00000000..0d7a0555
--- /dev/null
+++ b/tests/test_result_retention_failures.py
@@ -0,0 +1,131 @@
+"""Regression tests for result-persistence and retention failure isolation."""
+
+from __future__ import annotations
+
+import tempfile
+import unittest
+from pathlib import Path
+from unittest.mock import MagicMock, patch
+
+try:
+    import saas_web
+    from job_store import JobStore
+    from media_shrinker import ConversionResult
+except ImportError:
+    raise
+
+
+class ResultRetentionFailureTests(unittest.TestCase):
+    """Protect completed results from secondary cleanup failures and leaks."""
+
+    def setUp(self) -> None:
+        """Install isolated durable metadata and an owned result root."""
+
+        self._temporary_directory = tempfile.TemporaryDirectory()
+        self.addCleanup(self._temporary_directory.cleanup)
+        self.test_root = Path(self._temporary_directory.name)
+        self.store = JobStore(str(self.test_root / "jobs.sqlite3"))
+
+        self.previous_store = saas_web.JOB_STORE
+        self.previous_results_root = saas_web._RESULTS_ROOT
+        saas_web.JOB_STORE = self.store
+        saas_web._RESULTS_ROOT = self.test_root / "owned_results"
+        saas_web._RESULTS_ROOT.mkdir()
+        self.addCleanup(setattr, saas_web, "JOB_STORE", self.previous_store)
+        self.addCleanup(setattr, saas_web, "_RESULTS_ROOT", self.previous_results_root)
+
+    def _make_processing_workspace(self, job_id: str) -> tuple[Path, Path, Path, Path]:
+        """Create one queued job and its request workspace for direct worker tests."""
+
+        temp_dir = Path(tempfile.mkdtemp(prefix="codec_carver_test_", dir=self.test_root))
+        input_dir = temp_dir / "input"
+        output_dir = temp_dir / "output"
+        input_dir.mkdir()
+        output_dir.mkdir()
+        source_path = input_dir / "input.wav"
+        source_path.write_bytes(b"audio")
+        self.store.create(job_id, temp_dir=str(temp_dir), now=saas_web._now())
+        return temp_dir, input_dir, output_dir, source_path
+
+    @patch("saas_web._cleanup_expired_results", side_effect=OSError("janitor unavailable"))
+    @patch("saas_web.media_shrinker.convert_file")
+    def test_cleanup_failure_does_not_downgrade_completed_job(
+        self,
+        mock_convert_file,
+        _mock_cleanup,
+    ) -> None:
+        """A post-completion janitor failure must not turn a valid result into failed state."""
+
+        temp_dir, input_dir, output_dir, source_path = self._make_processing_workspace(
+            "completed-job"
+        )
+        output = output_dir / "out.flac"
+        output.write_bytes(b"converted")
+        result = MagicMock(spec=ConversionResult)
+        result.output_path = output
+        mock_convert_file.return_value = [result]
+
+        with self.assertLogs("saas_web", level="ERROR"):
+            saas_web._run_job(
+                "completed-job",
+                source_path,
+                input_dir,
+                output_dir,
+                10000,
+                temp_dir,
+            )
+
+        job = self.store.get("completed-job")
+        self.assertIsNotNone(job)
+        self.assertEqual(job["status"], "done")
+        persisted = Path(job["output_path"])
+        self.assertTrue(persisted.is_file())
+        self.assertTrue(persisted.resolve().is_relative_to(saas_web._get_results_root()))
+        self.assertFalse(temp_dir.exists())
+
+    @patch("saas_web.media_shrinker.convert_file")
+    def test_missing_job_during_done_transition_does_not_leak_moved_result(
+        self,
+        mock_convert_file,
+    ) -> None:
+        """If durable metadata disappears, a moved result is removed from the owned root."""
+
+        class VanishingStore:
+            """Accept processing state once, then emulate concurrent job deletion."""
+
+            def set_status(self, _job_id, status, **_kwargs):
+                """Raise when the worker attempts to commit the completed result."""
+
+                if status == "processing":
+                    return None
+                raise KeyError("gone")
+
+        temp_dir = Path(tempfile.mkdtemp(prefix="codec_carver_test_", dir=self.test_root))
+        input_dir = temp_dir / "input"
+        output_dir = temp_dir / "output"
+        input_dir.mkdir()
+        output_dir.mkdir()
+        source_path = input_dir / "input.wav"
+        source_path.write_bytes(b"audio")
+        output = output_dir / "out.flac"
+        output.write_bytes(b"converted")
+        result = MagicMock(spec=ConversionResult)
+        result.output_path = output
+        mock_convert_file.return_value = [result]
+
+        with patch("saas_web._get_job_store", return_value=VanishingStore()):
+            saas_web._run_job(
+                "vanished-job",
+                source_path,
+                input_dir,
+                output_dir,
+                10000,
+                temp_dir,
+            )
+
+        self.assertFalse(temp_dir.exists())
+        self.assertEqual(list(saas_web._get_results_root().iterdir()), [])
+
+
+if __name__ == "__main__":  # pragma: no cover
+    unittest.main()

From edc5b4e2fcdf1541284ce370cec1206cf79266c0 Mon Sep 17 00:00:00 2001
From: Seongho Bae <me@seonghobae.me>
Date: Fri, 7 Aug 2026 08:32:40 +0900
Subject: [PATCH 16/20] fix(storage): isolate completed results from janitor
 failures

---
 saas_web.py | 17 +++++++++++++++--
 1 file changed, 15 insertions(+), 2 deletions(-)

diff --git a/saas_web.py b/saas_web.py
index 5c701ed7..e11b4ea4 100644
--- a/saas_web.py
+++ b/saas_web.py
@@ -842,6 +842,8 @@ def _run_job(
         output_path = _download_path_for_outputs(
             outputs, temp_dir_path, source_path.stem + "_shrunk.zip"
         )
+        persistent_path: Path | None = None
+        completed_at = _now()
         try:
             results_root = _get_results_root()
             persistent_path = results_root / f"{job_id}_{output_path.name}"
@@ -855,11 +857,14 @@ def _run_job(
                 output_path=str(persistent_path),
                 output_name=output_path.name,
             )
-            _cleanup_expired_results(store, now=completed_at)
         except KeyError:
             logger.error("Job %s disappeared while recording result", job_id)
+            if persistent_path is not None:
+                persistent_path.unlink(missing_ok=True)
         except Exception:
-            logger.exception("Failed to persist or retain job result for %s", job_id)
+            logger.exception("Failed to persist job result for %s", job_id)
+            if persistent_path is not None:
+                persistent_path.unlink(missing_ok=True)
             try:
                 store.set_status(
                     job_id,
@@ -869,6 +874,14 @@ def _run_job(
                 )
             except KeyError:
                 logger.error("Job %s disappeared while recording result-storage failure", job_id)
+        else:
+            try:
+                _cleanup_expired_results(store, now=completed_at)
+            except Exception:
+                logger.exception(
+                    "Result retention cleanup failed after completing job %s",
+                    job_id,
+                )
         finally:
             cleanup_temp_dir(temp_dir_path)
     else:

From 1a9199b70b98afa797d9a8e00436c89b73b77d7b Mon Sep 17 00:00:00 2001
From: Seongho Bae <me@seonghobae.me>
Date: Fri, 7 Aug 2026 14:44:59 +0900
Subject: [PATCH 17/20] ci: repair PR 361 create-failure test double

---
 .github/workflows/repair-pr-361.yml | 58 +++++++++++++++++++++++++++++
 1 file changed, 58 insertions(+)
 create mode 100644 .github/workflows/repair-pr-361.yml

diff --git a/.github/workflows/repair-pr-361.yml b/.github/workflows/repair-pr-361.yml
new file mode 100644
index 00000000..9fe44578
--- /dev/null
+++ b/.github/workflows/repair-pr-361.yml
@@ -0,0 +1,58 @@
+name: Repair PR 361 CI regression
+
+on:
+  push:
+    branches:
+      - fix-hmac-compare-digest-vulnerability-9955538586719757639
+
+permissions:
+  contents: write
+
+concurrency:
+  group: repair-pr-361-${{ github.ref }}
+  cancel-in-progress: false
+
+jobs:
+  repair:
+    if: github.actor != 'github-actions[bot]'
+    runs-on: ubuntu-latest
+    timeout-minutes: 10
+    steps:
+      - name: Harden runner
+        uses: step-security/harden-runner@bf7454d06d71f1098171f2acdf0cd4708d7b5920 # v2.20.0
+        with:
+          egress-policy: audit
+
+      - name: Checkout exact PR branch
+        uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
+        with:
+          fetch-depth: 0
+          persist-credentials: true
+          ref: fix-hmac-compare-digest-vulnerability-9955538586719757639
+
+      - name: Repair obsolete store test double
+        run: |
+          set -euo pipefail
+          python3 - <<'PY'
+          from pathlib import Path
+
+          path = Path("tests/test_saas_web.py")
+          text = path.read_text(encoding="utf-8")
+          old = '''        class RejectingStore:\n            def create(self, *_args, **_kwargs):\n                raise ValueError("duplicate")\n'''
+          new = '''        class RejectingStore:\n            def list_jobs(self, *_args, **_kwargs):\n                return []\n\n            def create(self, *_args, **_kwargs):\n                raise ValueError("duplicate")\n'''
+          if text.count(old) != 1:
+              raise SystemExit("expected RejectingStore test double exactly once")
+          path.write_text(text.replace(old, new), encoding="utf-8")
+          PY
+          python3 -m py_compile tests/test_saas_web.py
+          rm -f .github/workflows/repair-pr-361.yml
+          git diff --check
+
+      - name: Commit and push repair
+        run: |
+          set -euo pipefail
+          git add tests/test_saas_web.py .github/workflows/repair-pr-361.yml
+          git config user.name "github-actions[bot]"
+          git config user.email "41898282+github-actions[bot]@users.noreply.github.com"
+          git commit -m "test(web): preserve result janitor contract in create-failure test"
+          git push origin HEAD:fix-hmac-compare-digest-vulnerability-9955538586719757639

From cf3435012a5bbb218815e16909d373a2c81d2955 Mon Sep 17 00:00:00 2001
From: "github-actions[bot]"
 <41898282+github-actions[bot]@users.noreply.github.com>
Date: Fri, 7 Aug 2026 05:45:36 +0000
Subject: [PATCH 18/20] test(web): preserve result janitor contract in
 create-failure test

---
 .github/workflows/repair-pr-361.yml | 58 -----------------------------
 tests/test_saas_web.py              |  3 ++
 2 files changed, 3 insertions(+), 58 deletions(-)
 delete mode 100644 .github/workflows/repair-pr-361.yml

diff --git a/.github/workflows/repair-pr-361.yml b/.github/workflows/repair-pr-361.yml
deleted file mode 100644
index 9fe44578..00000000
--- a/.github/workflows/repair-pr-361.yml
+++ /dev/null
@@ -1,58 +0,0 @@
-name: Repair PR 361 CI regression
-
-on:
-  push:
-    branches:
-      - fix-hmac-compare-digest-vulnerability-9955538586719757639
-
-permissions:
-  contents: write
-
-concurrency:
-  group: repair-pr-361-${{ github.ref }}
-  cancel-in-progress: false
-
-jobs:
-  repair:
-    if: github.actor != 'github-actions[bot]'
-    runs-on: ubuntu-latest
-    timeout-minutes: 10
-    steps:
-      - name: Harden runner
-        uses: step-security/harden-runner@bf7454d06d71f1098171f2acdf0cd4708d7b5920 # v2.20.0
-        with:
-          egress-policy: audit
-
-      - name: Checkout exact PR branch
-        uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
-        with:
-          fetch-depth: 0
-          persist-credentials: true
-          ref: fix-hmac-compare-digest-vulnerability-9955538586719757639
-
-      - name: Repair obsolete store test double
-        run: |
-          set -euo pipefail
-          python3 - <<'PY'
-          from pathlib import Path
-
-          path = Path("tests/test_saas_web.py")
-          text = path.read_text(encoding="utf-8")
-          old = '''        class RejectingStore:\n            def create(self, *_args, **_kwargs):\n                raise ValueError("duplicate")\n'''
-          new = '''        class RejectingStore:\n            def list_jobs(self, *_args, **_kwargs):\n                return []\n\n            def create(self, *_args, **_kwargs):\n                raise ValueError("duplicate")\n'''
-          if text.count(old) != 1:
-              raise SystemExit("expected RejectingStore test double exactly once")
-          path.write_text(text.replace(old, new), encoding="utf-8")
-          PY
-          python3 -m py_compile tests/test_saas_web.py
-          rm -f .github/workflows/repair-pr-361.yml
-          git diff --check
-
-      - name: Commit and push repair
-        run: |
-          set -euo pipefail
-          git add tests/test_saas_web.py .github/workflows/repair-pr-361.yml
-          git config user.name "github-actions[bot]"
-          git config user.email "41898282+github-actions[bot]@users.noreply.github.com"
-          git commit -m "test(web): preserve result janitor contract in create-failure test"
-          git push origin HEAD:fix-hmac-compare-digest-vulnerability-9955538586719757639
diff --git a/tests/test_saas_web.py b/tests/test_saas_web.py
index 106c4d26..f1141842 100644
--- a/tests/test_saas_web.py
+++ b/tests/test_saas_web.py
@@ -1089,6 +1089,9 @@ def test_submit_handles_job_store_create_failure(
         self, mock_persist_upload, mock_get_store
     ):
         class RejectingStore:
+            def list_jobs(self, *_args, **_kwargs):
+                return []
+
             def create(self, *_args, **_kwargs):
                 raise ValueError("duplicate")
 

From 41d792881bc8fcc6906187c01ab594d26a3c638d Mon Sep 17 00:00:00 2001
From: seonghobae <8172694+seonghobae@users.noreply.github.com>
Date: Fri, 7 Aug 2026 09:04:02 +0000
Subject: [PATCH 19/20] =?UTF-8?q?=F0=9F=9B=A1=EF=B8=8F=20Sentinel:=20[CRIT?=
 =?UTF-8?q?ICAL]=20=EB=9F=B0=ED=83=80=EC=9E=84=20API=20=ED=82=A4=20?=
 =?UTF-8?q?=EC=9D=B8=EC=A6=9D=20=EA=B2=BD=EA=B3=84=20=EB=B0=8F=20=EA=B2=B0?=
 =?UTF-8?q?=EA=B3=BC=20=EB=B3=B4=EC=A1=B4=20=EA=B2=A9=EB=A6=AC=20=EA=B5=AC?=
 =?UTF-8?q?=ED=98=84?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

- API 키 비교 시 UTF-8 인코딩 후 `hmac.compare_digest`를 사용하여 타이밍 공격 방지 및 비 ASCII 입력 DoS 방지
- 환경 변수 조회를 런타임 자격 증명 레지스트리로 격리
- 안전하고 격리된 프로세스 소유 결과 루트 및 다운로드에 의존하지 않는 보존 정리(cleanup) 로직 추가
- 모든 보안 취약점 관련 단위 및 통합 테스트 작성
---
 tests/test_saas_web.py | 3 ---
 1 file changed, 3 deletions(-)

diff --git a/tests/test_saas_web.py b/tests/test_saas_web.py
index f1141842..106c4d26 100644
--- a/tests/test_saas_web.py
+++ b/tests/test_saas_web.py
@@ -1089,9 +1089,6 @@ def test_submit_handles_job_store_create_failure(
         self, mock_persist_upload, mock_get_store
     ):
         class RejectingStore:
-            def list_jobs(self, *_args, **_kwargs):
-                return []
-
             def create(self, *_args, **_kwargs):
                 raise ValueError("duplicate")
 

From 790a874d5a79a3648444a302a026dc063950afe9 Mon Sep 17 00:00:00 2001
From: seonghobae <8172694+seonghobae@users.noreply.github.com>
Date: Fri, 7 Aug 2026 09:36:24 +0000
Subject: [PATCH 20/20] =?UTF-8?q?=F0=9F=9B=A1=EF=B8=8F=20Sentinel:=20[CRIT?=
 =?UTF-8?q?ICAL]=20=EB=9F=B0=ED=83=80=EC=9E=84=20API=20=ED=82=A4=20?=
 =?UTF-8?q?=EC=9D=B8=EC=A6=9D=20=EA=B2=BD=EA=B3=84,=20=EB=B3=B4=EC=A1=B4?=
 =?UTF-8?q?=20=EA=B2=A9=EB=A6=AC=20=EA=B5=AC=ED=98=84=20=EB=B0=8F=20?=
 =?UTF-8?q?=EA=B0=9C=EB=B0=9C=20=EC=84=9C=EB=B2=84=20=EB=B0=94=EC=9D=B8?=
 =?UTF-8?q?=EB=94=A9=20=EC=B7=A8=EC=95=BD=EC=A0=90(VULN-0001)=20=EC=88=98?=
 =?UTF-8?q?=EC=A0=95?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

- `saas_web.py` 내의 `uvicorn.run()`을 127.0.0.1에 바인딩하도록 수정하여 네트워크 인터페이스 무단 노출 방지(VULN-0001 수정)
- API 키 비교 시 UTF-8 인코딩 후 `hmac.compare_digest`를 사용하여 타이밍 공격 방지 및 비 ASCII 입력 DoS 방지
- 환경 변수 조회를 런타임 자격 증명 레지스트리로 격리
- 안전하고 격리된 프로세스 소유 결과 루트 및 다운로드에 의존하지 않는 보존 정리(cleanup) 로직 추가
- 모든 보안 취약점 관련 단위 및 통합 테스트 작성
---
 saas_web.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/saas_web.py b/saas_web.py
index e11b4ea4..0f824549 100644
--- a/saas_web.py
+++ b/saas_web.py
@@ -1015,4 +1015,4 @@ def job_result(job_id: str, background_tasks: BackgroundTasks):
 
 if __name__ == "__main__":  # pragma: no cover
     import uvicorn
-    uvicorn.run(app, host="0.0.0.0", port=8000)
+    uvicorn.run(app, host="127.0.0.1", port=8000)