Bound .comfy_environment read at 128 bytes; add unit tests

Defense-in-depth: cap readline() so a malformed or maliciously-large single-line file cannot blow up memory before the value is sanitized.

Adds tests-unit/deploy_environment_test.py covering: missing file fallback, basic read, whitespace strip, multi-line (only first line used), empty + whitespace-only files, control-char stripping (header-injection protection), non-ASCII stripping, 128-byte read cap, cache stickiness, and OSError fallback.

Amp-Thread-ID: https://ampcode.com/threads/T-019df26e-96f4-7518-94da-0e4263680e3c
Co-authored-by: Amp <amp@ampcode.com>

comfy/deploy_environment.py

@@ -19,7 +19,9 @@ def get_deploy_environment() -> str:
     env_file = os.path.join(folder_paths.base_path, _ENV_FILENAME)
     try:
         with open(env_file, encoding="utf-8") as f:
-            first_line = f.readline().strip()
+            # Cap the read so a malformed or maliciously crafted file (e.g.
+            # a single huge line with no newline) can't blow up memory.
+            first_line = f.readline(128).strip()
             value = "".join(c for c in first_line if 32 <= ord(c) < 127)
             if value:
                 _cached_value = value

tests-unit/deploy_environment_test.py

@@ -0,0 +1,84 @@
+"""Tests for comfy.deploy_environment."""
+
+import os
+
+import pytest
+
+from comfy import deploy_environment
+from comfy.deploy_environment import get_deploy_environment
+
+
+@pytest.fixture(autouse=True)
+def _reset_cache_and_base_path(tmp_path, monkeypatch):
+    """Reset the module cache and point folder_paths.base_path at a tmp dir for each test."""
+    monkeypatch.setattr(deploy_environment, "_cached_value", None)
+    import folder_paths
+    monkeypatch.setattr(folder_paths, "base_path", str(tmp_path))
+    yield
+    monkeypatch.setattr(deploy_environment, "_cached_value", None)
+
+
+def _write_env_file(tmp_path, content: str) -> str:
+    path = os.path.join(str(tmp_path), ".comfy_environment")
+    with open(path, "w", encoding="utf-8") as f:
+        f.write(content)
+    return path
+
+
+class TestGetDeployEnvironment:
+    def test_returns_local_git_when_file_missing(self):
+        assert get_deploy_environment() == "local_git"
+
+    def test_reads_value_from_file(self, tmp_path):
+        _write_env_file(tmp_path, "local_desktop2_standalone\n")
+        assert get_deploy_environment() == "local_desktop2_standalone"
+
+    def test_strips_trailing_whitespace_and_newline(self, tmp_path):
+        _write_env_file(tmp_path, "  local_desktop2_standalone  \n")
+        assert get_deploy_environment() == "local_desktop2_standalone"
+
+    def test_only_first_line_is_used(self, tmp_path):
+        _write_env_file(tmp_path, "first_line\nsecond_line\n")
+        assert get_deploy_environment() == "first_line"
+
+    def test_empty_file_falls_back_to_default(self, tmp_path):
+        _write_env_file(tmp_path, "")
+        assert get_deploy_environment() == "local_git"
+
+    def test_empty_after_whitespace_strip_falls_back_to_default(self, tmp_path):
+        _write_env_file(tmp_path, "   \n")
+        assert get_deploy_environment() == "local_git"
+
+    def test_strips_control_chars_within_first_line(self, tmp_path):
+        # Embedded NUL/control chars in the value should be stripped
+        # (header-injection / smuggling protection).
+        _write_env_file(tmp_path, "abc\x00\x07xyz\n")
+        assert get_deploy_environment() == "abcxyz"
+
+    def test_strips_non_ascii_characters(self, tmp_path):
+        _write_env_file(tmp_path, "café-é\n")
+        assert get_deploy_environment() == "caf-"
+
+    def test_caps_read_at_128_bytes(self, tmp_path):
+        # A single huge line with no newline must not be fully read into memory.
+        huge = "x" * 10_000
+        _write_env_file(tmp_path, huge)
+        result = get_deploy_environment()
+        assert result == "x" * 128
+
+    def test_result_is_cached_across_calls(self, tmp_path):
+        path = _write_env_file(tmp_path, "first_value\n")
+        assert get_deploy_environment() == "first_value"
+        # Overwrite the file — cached value should still be returned.
+        with open(path, "w", encoding="utf-8") as f:
+            f.write("second_value\n")
+        assert get_deploy_environment() == "first_value"
+
+    def test_unreadable_file_falls_back_to_default(self, tmp_path, monkeypatch):
+        _write_env_file(tmp_path, "should_not_be_used\n")
+
+        def _boom(*args, **kwargs):
+            raise OSError("simulated read failure")
+
+        monkeypatch.setattr("builtins.open", _boom)
+        assert get_deploy_environment() == "local_git"