fix(init): harden extension archive downloads

Signed-off-by: Asish Kumar <officialasishkumar@gmail.com>

Author: officialasishkumar

README.md

@@ -348,7 +348,7 @@ Community projects that extend, visualize, or build on Spec Kit:
 
 ## Available Slash Commands
 
-After running `specify init`, your AI coding agent will have access to these slash commands for structured development. If you pass `--ai <agent> --ai-skills`, Spec Kit installs agent skills instead of slash-command prompt files; `--ai-skills` requires `--ai`.
+After running `specify init --integration <agent>`, your AI coding agent will have access to these slash commands for structured development. Integrations choose the appropriate command format for each agent, including native skills where supported. The legacy `--ai` and `--ai-skills` options remain available as deprecated aliases.
 
 ### Core Commands
 

src/specify_cli/__init__.py

@@ -748,6 +748,10 @@ def _validate_extension_url(url: str) -> None:
         )
 
 
+MAX_EXTENSION_ARCHIVE_BYTES = 100 * 1024 * 1024
+DOWNLOAD_CHUNK_BYTES = 1024 * 1024
+
+
 def _install_extension_archive(
     manager: Any,
     archive_path: Path,
@@ -776,6 +780,10 @@ def _install_extension_archive(
                     raise ValidationError(
                         f"Unsafe link in TAR archive: {member.name}"
                     )
+                if not (member.isfile() or member.isdir()):
+                    raise ValidationError(
+                        f"Unsupported TAR member type in archive: {member.name}"
+                    )
                 member_path = (temp_path / member.name).resolve()
                 try:
                     member_path.relative_to(temp_path_resolved)
@@ -846,7 +854,36 @@ def _download_and_install_extension_url(
 
         try:
             with urllib.request.urlopen(source_url, timeout=60) as response:
-                archive_path.write_bytes(response.read())
+                content_length = response.headers.get("Content-Length")
+                try:
+                    content_length_bytes = (
+                        int(content_length) if content_length is not None else None
+                    )
+                except ValueError:
+                    content_length_bytes = None
+
+                if (
+                    content_length_bytes
+                    and content_length_bytes > MAX_EXTENSION_ARCHIVE_BYTES
+                ):
+                    raise ExtensionError(
+                        "Extension archive is too large; maximum allowed size "
+                        f"is {MAX_EXTENSION_ARCHIVE_BYTES // (1024 * 1024)} MiB."
+                    )
+
+                downloaded = 0
+                with archive_path.open("wb") as fh:
+                    while True:
+                        chunk = response.read(DOWNLOAD_CHUNK_BYTES)
+                        if not chunk:
+                            break
+                        downloaded += len(chunk)
+                        if downloaded > MAX_EXTENSION_ARCHIVE_BYTES:
+                            raise ExtensionError(
+                                "Extension archive is too large; maximum allowed "
+                                f"size is {MAX_EXTENSION_ARCHIVE_BYTES // (1024 * 1024)} MiB."
+                            )
+                        fh.write(chunk)
         except urllib.error.URLError as exc:
             raise ExtensionError(
                 f"Failed to download extension from {display_url}: {exc}"

tests/integrations/test_cli.py

@@ -2,7 +2,10 @@
 
 import json
 import os
+import tarfile
+from io import BytesIO
 
+import pytest
 import yaml
 
 from tests.conftest import strip_ansi
@@ -338,15 +341,27 @@ def test_extension_url_output_redacts_credentials_and_query(self, tmp_path, monk
         import specify_cli
         from specify_cli import app
 
+        payload = b"fake-archive"
+
         class FakeResponse:
+            headers = {"Content-Length": str(len(payload))}
+
+            def __init__(self):
+                self.offset = 0
+
             def __enter__(self):
                 return self
 
             def __exit__(self, exc_type, exc, tb):
                 return False
 
             def read(self, size=-1):
-                return b"fake-archive"
+                if self.offset >= len(payload):
+                    return b""
+                end = min(self.offset + size, len(payload))
+                chunk = payload[self.offset:end]
+                self.offset = end
+                return chunk
 
         def fake_urlopen(*args, **kwargs):
             return FakeResponse()
@@ -373,6 +388,82 @@ def fake_urlopen(*args, **kwargs):
         assert "user:secret" not in result.output
         assert "token=abc123" not in result.output
 
+    def test_tar_extension_archive_rejects_special_members(self, tmp_path):
+        """TAR extension archives reject non-file and non-directory members."""
+        from specify_cli import _install_extension_archive
+        from specify_cli.extensions import ValidationError
+
+        archive_path = tmp_path / "unsafe-extension.tar"
+        manifest = b"extension:\n  id: test-ext\n  name: Test\n  version: 1.0.0\n"
+
+        with tarfile.open(archive_path, "w") as tf:
+            manifest_info = tarfile.TarInfo("extension.yml")
+            manifest_info.size = len(manifest)
+            tf.addfile(manifest_info, BytesIO(manifest))
+
+            fifo_info = tarfile.TarInfo("unsafe-fifo")
+            fifo_info.type = tarfile.FIFOTYPE
+            tf.addfile(fifo_info)
+
+        with pytest.raises(ValidationError, match="Unsupported TAR member type"):
+            _install_extension_archive(object(), archive_path, "0.0.0")
+
+    def test_extension_url_downloads_in_bounded_chunks(self, tmp_path, monkeypatch):
+        """URL extension downloads stream to disk instead of reading all bytes."""
+        import urllib.request
+        import specify_cli
+
+        payload = b"archive-bytes"
+        read_sizes = []
+
+        class FakeResponse:
+            headers = {"Content-Length": str(len(payload))}
+
+            def __init__(self):
+                self.offset = 0
+
+            def __enter__(self):
+                return self
+
+            def __exit__(self, exc_type, exc, tb):
+                return False
+
+            def read(self, size=-1):
+                read_sizes.append(size)
+                if self.offset >= len(payload):
+                    return b""
+                end = min(self.offset + size, len(payload))
+                chunk = payload[self.offset:end]
+                self.offset = end
+                return chunk
+
+        def fake_urlopen(url, timeout):
+            assert url == "https://example.com/extension.zip"
+            assert timeout == 60
+            return FakeResponse()
+
+        def fake_install(manager, archive_path, speckit_version, priority=10):
+            assert archive_path.read_bytes() == payload
+            assert speckit_version == "0.0.0"
+            assert priority == 10
+            return "installed"
+
+        monkeypatch.setattr(urllib.request, "urlopen", fake_urlopen)
+        monkeypatch.setattr(specify_cli, "_install_extension_archive", fake_install)
+
+        result = specify_cli._download_and_install_extension_url(
+            object(),
+            tmp_path,
+            "https://example.com/extension.zip",
+            "0.0.0",
+        )
+
+        assert result == "installed"
+        assert read_sizes == [
+            specify_cli.DOWNLOAD_CHUNK_BYTES,
+            specify_cli.DOWNLOAD_CHUNK_BYTES,
+        ]
+
 
 class TestForceExistingDirectory:
     """Tests for --force merging into an existing named directory."""