Address remaining security review feedback

Author: PascalThuet

.github/scripts/check_security_requirements.py

@@ -0,0 +1,101 @@
+"""Check that committed security audit requirements are up to date."""
+
+from __future__ import annotations
+
+import os
+import subprocess
+import sys
+from pathlib import Path
+
+
+REPO_ROOT = Path(__file__).resolve().parents[2]
+COMMITTED_REQUIREMENTS = REPO_ROOT / ".github" / "security-audit-requirements.txt"
+DEPENDENCY_INPUTS = ("pyproject.toml", ".github/security-audit-requirements.txt")
+
+
+def _dependency_diff_refs() -> tuple[str, str]:
+    base_ref = os.environ.get("DEPENDENCY_DIFF_BASE", "").strip()
+    head_ref = os.environ.get("DEPENDENCY_DIFF_HEAD", "").strip() or "HEAD"
+    if base_ref and not set(base_ref) <= {"0"}:
+        return base_ref, head_ref
+    return "HEAD^", "HEAD"
+
+
+def _dependency_inputs_changed() -> bool:
+    base_ref, head_ref = _dependency_diff_refs()
+    try:
+        result = subprocess.run(
+            [
+                "git",
+                "diff",
+                "--name-only",
+                base_ref,
+                head_ref,
+                "--",
+                *DEPENDENCY_INPUTS,
+            ],
+            check=True,
+            cwd=REPO_ROOT,
+            stderr=subprocess.PIPE,
+            stdout=subprocess.PIPE,
+            text=True,
+        )
+    except subprocess.CalledProcessError as exc:
+        print(
+            "Could not determine changed dependency inputs; checking requirements.",
+            file=sys.stderr,
+        )
+        if exc.stderr:
+            print(exc.stderr.strip(), file=sys.stderr)
+        return True
+
+    changed_inputs = [line for line in result.stdout.splitlines() if line]
+    if not changed_inputs:
+        print("Dependency audit inputs unchanged; sync check skipped.")
+        return False
+
+    print(f"Dependency audit inputs changed: {', '.join(changed_inputs)}")
+    return True
+
+
+def main() -> int:
+    if not _dependency_inputs_changed():
+        return 0
+
+    generated_requirements = Path(os.environ["GENERATED_REQUIREMENTS"])
+    generated_requirements.parent.mkdir(parents=True, exist_ok=True)
+
+    subprocess.run(
+        [
+            "uv",
+            "pip",
+            "compile",
+            "pyproject.toml",
+            "--extra",
+            "test",
+            "--universal",
+            "--generate-hashes",
+            "--quiet",
+            "--no-header",
+            "--output-file",
+            str(generated_requirements),
+        ],
+        check=True,
+        cwd=REPO_ROOT,
+    )
+
+    committed = COMMITTED_REQUIREMENTS.read_text(encoding="utf-8")
+    generated = generated_requirements.read_text(encoding="utf-8")
+    if committed == generated:
+        return 0
+
+    print(
+        "Regenerate .github/security-audit-requirements.txt with the documented "
+        "uv pip compile command.",
+        file=sys.stderr,
+    )
+    return 1
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

.github/security-audit-requirements.txt

@@ -1,5 +1,3 @@
-# This file was autogenerated by uv via the following command:
-#    uv pip compile pyproject.toml --extra test --universal --generate-hashes --output-file .github/security-audit-requirements.txt
 annotated-doc==0.0.4 \
     --hash=sha256:571ac1dc6991c450b25a9c2d84a3705e2ae7a53467b5d111c24fa8baabbed320 \
     --hash=sha256:fbcda96e87e9c92ad167c2e53839e57503ecfda18804ea28102353485033faa4

.github/workflows/security.yml

@@ -23,6 +23,8 @@ jobs:
     steps:
       - name: Checkout
         uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6
+        with:
+          fetch-depth: 2
 
       - name: Install uv
         uses: astral-sh/setup-uv@08807647e7069bb48b6ef5acd8ec9567f424441b # v8.1.0
@@ -41,6 +43,14 @@ jobs:
         if: ${{ github.event_name == 'schedule' }}
         run: uvx --from pip-audit==2.10.0 pip-audit --disable-pip --require-hashes -r "${{ runner.temp }}/spec-kit-audit-requirements.txt" --progress-spinner off
 
+      - name: Check committed audit requirements are current
+        if: ${{ github.event_name != 'schedule' }}
+        env:
+          DEPENDENCY_DIFF_BASE: ${{ github.event.pull_request.base.sha || github.event.before || '' }}
+          DEPENDENCY_DIFF_HEAD: ${{ github.sha }}
+          GENERATED_REQUIREMENTS: ${{ runner.temp }}/security-audit-requirements.txt
+        run: python .github/scripts/check_security_requirements.py
+
       - name: Run pip-audit (committed requirements)
         if: ${{ github.event_name != 'schedule' }}
         run: uvx --from pip-audit==2.10.0 pip-audit --disable-pip --require-hashes -r .github/security-audit-requirements.txt --progress-spinner off

CONTRIBUTING.md

@@ -91,7 +91,7 @@ uvx --from bandit==1.9.4 bandit -r src -lll --baseline .github/bandit-baseline.j
 Run these before changing dependency metadata, workflow execution code, subprocess usage, or security-sensitive paths. Pull request, push, and manual CI audits use the committed hashed requirements file so they stay deterministic. The scheduled CI audit also resolves the runtime and `test` extra dependency set across the supported Python and OS matrix to catch newly published advisories. If dependency metadata changes, refresh the committed audit input before running pip-audit:
 
 ```bash
-uv pip compile pyproject.toml --extra test --universal --generate-hashes --quiet --output-file .github/security-audit-requirements.txt
+uv pip compile pyproject.toml --extra test --universal --generate-hashes --quiet --no-header --output-file .github/security-audit-requirements.txt
 ```
 
 ### Manual testing

src/specify_cli/__init__.py

@@ -54,6 +54,7 @@
 from rich.tree import Tree
 from typer.core import TyperGroup
 
+from ._download_security import read_response_limited
 from .integration_runtime import (
     invoke_separator_for_integration as _invoke_separator_for_integration,
     resolve_integration_options as _resolve_integration_options_impl,
@@ -1772,7 +1773,13 @@ def _fetch_latest_release_tag() -> tuple[str | None, str | None]:
         req.add_header("Authorization", f"Bearer {token}")
     try:
         with urllib.request.urlopen(req, timeout=5) as resp:
-            payload = json.loads(resp.read().decode("utf-8"))
+            payload = json.loads(
+                read_response_limited(
+                    resp,
+                    max_bytes=1024 * 1024,
+                    label="GitHub latest release",
+                ).decode("utf-8")
+            )
             tag = payload.get("tag_name")
             if not isinstance(tag, str) or not tag:
                 raise ValueError("GitHub API response missing valid tag_name")
@@ -3376,8 +3383,10 @@ def preset_add(
                 zip_path = Path(tmpdir) / "preset.zip"
                 try:
                     with urllib.request.urlopen(from_url, timeout=60) as response:
-                        zip_path.write_bytes(response.read())
-                except urllib.error.URLError as e:
+                        zip_path.write_bytes(
+                            read_response_limited(response, label=f"preset {from_url}")
+                        )
+                except (urllib.error.URLError, ValueError) as e:
                     console.print(f"[red]Error:[/red] Failed to download: {e}")
                     raise typer.Exit(1)
 
@@ -4280,12 +4289,15 @@ def extension_add(
 
                 try:
                     with urllib.request.urlopen(from_url, timeout=60) as response:
-                        zip_data = response.read()
+                        zip_data = read_response_limited(
+                            response,
+                            label=f"extension {from_url}",
+                        )
                     zip_path.write_bytes(zip_data)
 
                     # Install from downloaded ZIP
                     manifest = manager.install_from_zip(zip_path, speckit_version, priority=priority)
-                except urllib.error.URLError as e:
+                except (urllib.error.URLError, ValueError) as e:
                     console.print(f"[red]Error:[/red] Failed to download from {from_url}: {e}")
                     raise typer.Exit(1)
                 finally:
@@ -5526,7 +5538,7 @@ def _validate_and_install_local(yaml_path: Path, source_label: str) -> None:
                     console.print(f"[red]Error:[/red] URL redirected to non-HTTPS: {final_url}")
                     raise typer.Exit(1)
                 with tempfile.NamedTemporaryFile(suffix=".yml", delete=False) as tmp:
-                    tmp.write(resp.read())
+                    tmp.write(read_response_limited(resp, label=f"workflow {source}"))
                     tmp_path = Path(tmp.name)
         except typer.Exit:
             raise
@@ -5630,7 +5642,9 @@ def _validate_and_install_local(yaml_path: Path, source_label: str) -> None:
                     f"[red]Error:[/red] Workflow '{source}' redirected to non-HTTPS URL: {final_url}"
                 )
                 raise typer.Exit(1)
-            workflow_file.write_bytes(response.read())
+            workflow_file.write_bytes(
+                read_response_limited(response, label=f"workflow {source}")
+            )
     except Exception as exc:
         if workflow_dir.exists():
             import shutil

src/specify_cli/_download_security.py

@@ -80,7 +80,7 @@ def _safe_zip_name(name: str, *, error_type: type[ErrorT]) -> str:
 
     normalized = name.replace("\\", "/")
     path = PurePosixPath(normalized)
-    has_windows_drive = re.match(r"^[A-Za-z]:/", normalized) is not None
+    has_windows_drive = re.match(r"^[A-Za-z]:", normalized) is not None
     if (
         not path.parts
         or path.is_absolute()

src/specify_cli/extensions.py

@@ -250,7 +250,7 @@ def _validate(self):
 
             normalized_file = cmd["file"].replace("\\", "/")
             file_path = PurePosixPath(normalized_file)
-            has_windows_drive = re.match(r"^[A-Za-z]:/", normalized_file) is not None
+            has_windows_drive = re.match(r"^[A-Za-z]:", normalized_file) is not None
             if (
                 file_path.is_absolute()
                 or has_windows_drive
@@ -1921,7 +1921,13 @@ def _fetch_single_catalog(self, entry: CatalogEntry, force_refresh: bool = False
         # Fetch from network
         try:
             with self._open_url(entry.url, timeout=10) as response:
-                catalog_data = json.loads(response.read())
+                catalog_data = json.loads(
+                    read_response_limited(
+                        response,
+                        error_type=ExtensionError,
+                        label=f"extension catalog {entry.url}",
+                    )
+                )
 
             if "schema_version" not in catalog_data or "extensions" not in catalog_data:
                 raise ExtensionError(f"Invalid catalog format from {entry.url}")
@@ -2037,7 +2043,13 @@ def fetch_catalog(self, force_refresh: bool = False) -> Dict[str, Any]:
             import urllib.error
 
             with self._open_url(catalog_url, timeout=10) as response:
-                catalog_data = json.loads(response.read())
+                catalog_data = json.loads(
+                    read_response_limited(
+                        response,
+                        error_type=ExtensionError,
+                        label=f"extension catalog {catalog_url}",
+                    )
+                )
 
             # Validate catalog structure
             if "schema_version" not in catalog_data or "extensions" not in catalog_data:

src/specify_cli/integrations/catalog.py

@@ -21,6 +21,8 @@
 import yaml
 from packaging import version as pkg_version
 
+from .._download_security import read_response_limited
+
 
 # ---------------------------------------------------------------------------
 # Errors
@@ -294,7 +296,13 @@ def _fetch_single_catalog(
                 final_url = resp.geturl()
                 if final_url != entry.url:
                     self._validate_catalog_url(final_url)
-                catalog_data = json.loads(resp.read())
+                catalog_data = json.loads(
+                    read_response_limited(
+                        resp,
+                        error_type=IntegrationCatalogError,
+                        label=f"integration catalog {entry.url}",
+                    )
+                )
 
             if not isinstance(catalog_data, dict):
                 raise IntegrationCatalogError(

src/specify_cli/presets.py

@@ -226,7 +226,7 @@ def _validate(self):
                 )
             normalized = file_path.replace("\\", "/")
             normalized_path = PurePosixPath(normalized)
-            has_windows_drive = re.match(r"^[A-Za-z]:/", normalized) is not None
+            has_windows_drive = re.match(r"^[A-Za-z]:", normalized) is not None
             if normalized_path.is_absolute() or any(
                 part == ".." for part in normalized_path.parts
             ) or has_windows_drive:
@@ -2045,7 +2045,13 @@ def _fetch_single_catalog(self, entry: PresetCatalogEntry, force_refresh: bool =
 
         try:
             with self._open_url(entry.url, timeout=10) as response:
-                catalog_data = json.loads(response.read())
+                catalog_data = json.loads(
+                    read_response_limited(
+                        response,
+                        error_type=PresetError,
+                        label=f"preset catalog {entry.url}",
+                    )
+                )
 
             if (
                 "schema_version" not in catalog_data
@@ -2138,7 +2144,13 @@ def fetch_catalog(self, force_refresh: bool = False) -> Dict[str, Any]:
 
         try:
             with self._open_url(catalog_url, timeout=10) as response:
-                catalog_data = json.loads(response.read())
+                catalog_data = json.loads(
+                    read_response_limited(
+                        response,
+                        error_type=PresetError,
+                        label=f"preset catalog {catalog_url}",
+                    )
+                )
 
             if (
                 "schema_version" not in catalog_data

tests/integrations/test_integration_catalog.py

@@ -173,7 +173,7 @@ def __init__(self, data, url=""):
                 self._data = json.dumps(data).encode()
                 self._url = url
 
-            def read(self):
+            def read(self, _size=-1):
                 return self._data
 
             def geturl(self):
@@ -294,6 +294,50 @@ def test_invalid_catalog_format(self, tmp_path, monkeypatch):
         with pytest.raises(IntegrationCatalogError, match="Failed to fetch any integration catalog"):
             cat.search()
 
+    def test_fetch_single_catalog_uses_bounded_read(self, tmp_path, monkeypatch):
+        cat = IntegrationCatalog(tmp_path)
+        entry = IntegrationCatalogEntry(
+            url="https://example.com/catalog.json",
+            name="test",
+            priority=1,
+            install_allowed=True,
+        )
+
+        class FakeResponse:
+            def read(self, _size=-1):
+                return b"{}"
+
+            def geturl(self):
+                return entry.url
+
+            def __enter__(self):
+                return self
+
+            def __exit__(self, *_args):
+                pass
+
+        def fake_urlopen(url, timeout=10):
+            assert url == entry.url
+            assert timeout == 10
+            return FakeResponse()
+
+        def fake_read_response_limited(response, **kwargs):
+            assert isinstance(response, FakeResponse)
+            assert kwargs["error_type"] is IntegrationCatalogError
+            assert kwargs["label"] == "integration catalog https://example.com/catalog.json"
+            raise IntegrationCatalogError("catalog too large")
+
+        import urllib.request
+
+        monkeypatch.setattr(urllib.request, "urlopen", fake_urlopen)
+        monkeypatch.setattr(
+            "specify_cli.integrations.catalog.read_response_limited",
+            fake_read_response_limited,
+        )
+
+        with pytest.raises(IntegrationCatalogError, match="catalog too large"):
+            cat._fetch_single_catalog(entry, force_refresh=True)
+
     def test_clear_cache(self, tmp_path):
         (tmp_path / ".specify").mkdir()
         cat = IntegrationCatalog(tmp_path)
@@ -492,7 +536,7 @@ class FakeResponse:
             def __init__(self, data, url=""):
                 self._data = json.dumps(data).encode()
                 self._url = url
-            def read(self):
+            def read(self, _size=-1):
                 return self._data
             def geturl(self):
                 return self._url