fix: make artifact-download availability follow the snippet execution seam

Author: yyyyyyyan

app/sep/apps/snippets/app.py

@@ -33,6 +33,12 @@
 the library (``app.sep.snippets.celery``), not this package, so the app owns no
 Celery module to prefix it with and its schedule is seeded unconditionally by
 ``get_system_periodic_tasks``.
+
+Snippets declares no ``artifact_base_dirs``: the snippet artifact type follows
+snippet execution, which is library-owned, so it is declared statically in
+``app.sep.artifact_constants`` and resolves whether or not this app is
+activated. Declaring it here as well would trip the duplicate-type guard in
+``collect_base_dirs``.
 """
 
 from app.sep.apps.framework.apps import TaskExecutionApp
@@ -43,7 +49,6 @@
     maintenance_router,
 )
 from app.sep.snippets.config import snippets_settings
-from app.sep.snippets.constants import ARTIFACT_TYPE_SNIPPET
 from app.sep.snippets.crud import SnippetManager
 from app.sep.snippets.models.responses import SnippetsCapabilitiesResponse
 from app.sep.snippets.script_source import snippet_source
@@ -77,5 +82,4 @@ def _snippets_capabilities_provider() -> SnippetsCapabilitiesResponse:
     list_query_spec=SnippetManager.list_query_spec,
     capabilities_provider=_snippets_capabilities_provider,
     extra_routes=(approval_router, maintenance_router, artifact_router),
-    artifact_base_dirs={ARTIFACT_TYPE_SNIPPET: lambda: snippets_settings.SNIPPETS_DIR},
 )

app/sep/artifact_constants.py

@@ -13,13 +13,30 @@
 # You should have received a copy of the GNU Affero General Public License
 # along with this program. If not, see <https://www.gnu.org/licenses/>.
 
-"""Define constant for signing artifact-download URLs.
+"""Define the constants the artifact-download surface is built from.
 
 Houses the itsdangerous salt shared by the generic ``app.sep.routes.artifacts``
 route and the framework signer in ``app.sep.apps.framework.script_helpers``, so
-both sign and verify download tokens under the same namespace.
+both sign and verify download tokens under the same namespace, plus the
+base-dir declarations that are not owned by any activatable app.
 """
 
-__all__ = ["ARTIFACT_DOWNLOAD_SALT"]
+from collections.abc import Callable, Mapping
+from pathlib import Path
+
+from app.sep.snippets.config import snippets_settings
+from app.sep.snippets.constants import ARTIFACT_TYPE_SNIPPET
+
+__all__ = ["ARTIFACT_DOWNLOAD_SALT", "STATIC_ARTIFACT_BASE_DIRS"]
 
 ARTIFACT_DOWNLOAD_SALT = "artifact-download"
+
+#: Artifact base dirs seeding the download map. Not owned by a ``SEP.APPS`` app,
+#: so they are not registry-derived; ``collect_base_dirs`` seeds them ahead of
+#: the per-app declarations. The snippet directory is declared here because
+#: snippet execution is library-owned: ATW builds signed snippet-download URLs
+#: through ``app.sep.snippets.script_source`` whether or not the snippets app is
+#: activated, so the type it names must resolve on the same terms.
+STATIC_ARTIFACT_BASE_DIRS: Mapping[str, Callable[[], Path]] = {
+    ARTIFACT_TYPE_SNIPPET: lambda: snippets_settings.SNIPPETS_DIR,
+}

app/sep/main.py

@@ -50,6 +50,7 @@
 from app.sep.config import sep_settings, warn_if_base_url_lacks_root_path
 from app.sep.db import get_async_session_maker
 from app.sep.db.seed import get_system_periodic_tasks, init_sep_db
+from app.sep.routes.artifacts import router as artifacts_router
 from app.sep.settings_override import (
     build_sep_override_proxies,
     invalidate_pmm_clients,
@@ -342,10 +343,7 @@ async def sep_lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
     sep_app.include_router(download_files_router, prefix="/files")
     sep_app.include_router(execution_events_router, prefix="/execution-events")
 
-if any(app.artifact_base_dirs for app in get_app_registry()):
-    from app.sep.routes.artifacts import router as artifacts_router
-
-    sep_app.include_router(artifacts_router, prefix="/artifacts")
+sep_app.include_router(artifacts_router, prefix="/artifacts")
 
 sep_app.include_router(api_router)
 sep_app.include_router(top_level_api_router, include_in_schema=False)

app/sep/routes/artifacts.py

@@ -25,25 +25,31 @@
 from app.core.exceptions import HTTPBadRequestException, HTTPNotFoundException
 from app.core.security import crypto_timestamp_serializer
 from app.sep.apps.framework.registry import get_app_registry
-from app.sep.artifact_constants import ARTIFACT_DOWNLOAD_SALT
+from app.sep.artifact_constants import ARTIFACT_DOWNLOAD_SALT, STATIC_ARTIFACT_BASE_DIRS
 from app.sep.config import sep_settings
 
 router = APIRouter(include_in_schema=False)
 
 
 def collect_base_dirs() -> dict[str, Callable[[], Path]]:
-    """Collect every app's ``artifact_base_dirs`` into a single lookup map.
+    """Collect the static and per-app ``artifact_base_dirs`` into one lookup map.
+
+    Seeded from :data:`~app.sep.artifact_constants.STATIC_ARTIFACT_BASE_DIRS`
+    so a type whose producer is library-owned rather than app-owned resolves
+    regardless of which apps an image activates, then overlaid with every
+    registered app's own declaration.
 
     :return: A mapping from artifact-type discriminator to its base-dir thunk.
-    :raises ValueError: If two apps declare the same artifact type, which
-        would ambiguously route one app's downloads into another's directory.
+    :raises ValueError: If an artifact type is declared more than once — by two
+        apps, or by an app that re-declares a static type — which would
+        ambiguously route one producer's downloads into another's directory.
     """
-    base_dirs = {}
+    base_dirs = dict(STATIC_ARTIFACT_BASE_DIRS)
     for app in get_app_registry():
         for artifact_type, thunk in app.artifact_base_dirs.items():
             if artifact_type in base_dirs:
                 raise ValueError(
-                    f"Artifact type {artifact_type!r} declared by more than one app"
+                    f"Artifact type {artifact_type!r} declared more than once"
                 )
             base_dirs[artifact_type] = thunk
     return base_dirs

tests/app/sep/conftest.py

@@ -61,7 +61,7 @@
 ALEMBIC_INI = REPO_ROOT / "alembic.ini"
 
 REDUCED_ACTIVATION = [
-    App(module_name=name) for name in ("inventory", "snippets", "atw", "mysql_backups")
+    App(module_name=name) for name in ("inventory", "atw", "mysql_backups")
 ]
 """The PMM-embedded side-car activation list (``sidecar/settings.yaml``)."""
 

tests/app/sep/routes/test_artifacts.py

@@ -28,8 +28,9 @@
 
 from app.core.security import crypto_timestamp_serializer
 from app.sep.apps.framework.base import BaseApp
-from app.sep.artifact_constants import ARTIFACT_DOWNLOAD_SALT
+from app.sep.artifact_constants import ARTIFACT_DOWNLOAD_SALT, STATIC_ARTIFACT_BASE_DIRS
 from app.sep.routes.artifacts import collect_base_dirs
+from app.sep.snippets.constants import ARTIFACT_TYPE_SNIPPET
 
 
 def _make_token(payload: dict, salt: str = ARTIFACT_DOWNLOAD_SALT) -> str:
@@ -60,29 +61,69 @@ def test_raises_on_duplicate_artifact_type(self, mocker) -> None:
             collect_base_dirs()
 
     def test_flattens_distinct_artifact_types(self, mocker) -> None:
-        """Merge distinct per-app declarations into one map."""
-        snippet_dir = Path("/tmp/snippets")
+        """Merge distinct per-app declarations over the static seed."""
         dipper_dir = Path("/tmp/dipper")
-        first = BaseApp(
-            name="snippets",
-            uri_path="/snippets",
-            artifact_base_dirs={"snippet": lambda: snippet_dir},
-        )
-        second = BaseApp(
+        other_dir = Path("/tmp/other")
+        dipper = BaseApp(
             name="dipper",
             uri_path="/dipper",
             artifact_base_dirs={"dipper": lambda: dipper_dir},
         )
+        other = BaseApp(
+            name="other",
+            uri_path="/other",
+            artifact_base_dirs={"other": lambda: other_dir},
+        )
         mocker.patch(
             "app.sep.routes.artifacts.get_app_registry",
-            return_value=[first, second],
+            return_value=[dipper, other],
         )
 
         result = collect_base_dirs()
 
-        assert result.keys() == {"snippet", "dipper"}
-        assert result["snippet"]() == snippet_dir
+        assert result.keys() == {ARTIFACT_TYPE_SNIPPET, "dipper", "other"}
         assert result["dipper"]() == dipper_dir
+        assert result["other"]() == other_dir
+
+    def test_seeds_the_static_snippet_type_without_the_snippets_app(
+        self, mocker
+    ) -> None:
+        """Resolve the snippet type from the static map, with no app declaring it."""
+        mocker.patch(
+            "app.sep.routes.artifacts.get_app_registry",
+            return_value=[BaseApp(name="atw", uri_path="/atw")],
+        )
+
+        assert ARTIFACT_TYPE_SNIPPET in collect_base_dirs()
+
+    def test_raises_when_an_app_redeclares_a_static_type(self, mocker) -> None:
+        """Reject an app that re-declares a statically registered type."""
+        redeclaring = BaseApp(
+            name="snippets",
+            uri_path="/snippets",
+            artifact_base_dirs={ARTIFACT_TYPE_SNIPPET: lambda: Path("/tmp/other")},
+        )
+        mocker.patch(
+            "app.sep.routes.artifacts.get_app_registry", return_value=[redeclaring]
+        )
+
+        with pytest.raises(ValueError, match=ARTIFACT_TYPE_SNIPPET):
+            collect_base_dirs()
+
+    def test_does_not_mutate_the_static_map(self, mocker) -> None:
+        """Return a fresh dict so an app declaration cannot leak into the constant."""
+        declaring = BaseApp(
+            name="dipper",
+            uri_path="/dipper",
+            artifact_base_dirs={"dipper": lambda: Path("/tmp/dipper")},
+        )
+        mocker.patch(
+            "app.sep.routes.artifacts.get_app_registry", return_value=[declaring]
+        )
+
+        collect_base_dirs()
+
+        assert "dipper" not in STATIC_ARTIFACT_BASE_DIRS
 
 
 class TestDownloadArtifact:
@@ -100,9 +141,7 @@ def test_valid_snippet_token_existing_file_returns_200(self, test_client, tmp_pa
         }
         token = _make_token(payload)
 
-        with patch(
-            "app.sep.apps.snippets.app.snippets_settings.SNIPPETS_DIR", tmp_path
-        ):
+        with patch("app.sep.snippets.config.snippets_settings.SNIPPETS_DIR", tmp_path):
             response = test_client.get(f"/artifacts/download/{token}")
 
         assert response.status_code == HTTP_200_OK
@@ -143,7 +182,7 @@ def test_expired_token_returns_400(self, test_client, tmp_path):
 
         with (
             patch("app.sep.routes.artifacts.sep_settings.ARTIFACT_DOWNLOAD_TTL", 60),
-            patch("app.sep.apps.snippets.app.snippets_settings.SNIPPETS_DIR", tmp_path),
+            patch("app.sep.snippets.config.snippets_settings.SNIPPETS_DIR", tmp_path),
         ):
             response = test_client.get(
                 f"/artifacts/download/{token}", follow_redirects=False
@@ -176,9 +215,7 @@ def test_valid_token_nonexistent_file_returns_404(self, test_client, tmp_path):
         }
         token = _make_token(payload)
 
-        with patch(
-            "app.sep.apps.snippets.app.snippets_settings.SNIPPETS_DIR", tmp_path
-        ):
+        with patch("app.sep.snippets.config.snippets_settings.SNIPPETS_DIR", tmp_path):
             response = test_client.get(
                 f"/artifacts/download/{token}", follow_redirects=False
             )
@@ -194,9 +231,7 @@ def test_path_traversal_in_filename_returns_400(self, test_client, tmp_path):
         }
         token = _make_token(payload)
 
-        with patch(
-            "app.sep.apps.snippets.app.snippets_settings.SNIPPETS_DIR", tmp_path
-        ):
+        with patch("app.sep.snippets.config.snippets_settings.SNIPPETS_DIR", tmp_path):
             response = test_client.get(
                 f"/artifacts/download/{token}", follow_redirects=False
             )

tests/app/sep/test_main.py

@@ -18,7 +18,7 @@
 import importlib
 import logging
 from contextlib import asynccontextmanager, contextmanager
-from unittest.mock import AsyncMock, Mock
+from unittest.mock import AsyncMock, Mock, patch
 
 import pytest
 from fastapi import FastAPI, HTTPException, status
@@ -27,8 +27,10 @@
 from starlette.datastructures import URL
 
 import app.sep.main as main_module
+import app.sep.routes.artifacts as artifacts_module
 from app.core.alerts.config import alert_settings, AlertSettings
 from app.core.auth.exceptions import BaseAuthProviderException
+from app.core.security import crypto_timestamp_serializer
 from app.core.settings_override.lifecycle import ProxyEntry
 from app.core.settings_override.models import SettingClassEnum
 from app.sep.api.router import apps_router
@@ -38,6 +40,7 @@
     AppRegistry,
     get_app_registry,
 )
+from app.sep.artifact_constants import ARTIFACT_DOWNLOAD_SALT
 from app.sep.config import App, sep_settings, SEPSettings
 from app.sep.deps import get_session, PROTECTED_APP_KEYS
 from app.sep.main import lifespan as sep_module_lifespan
@@ -49,6 +52,7 @@
 )
 from app.sep.models import AppLifecycleEnum, AppState
 from app.sep.snippets.config import snippets_settings
+from app.sep.snippets.constants import ARTIFACT_TYPE_SNIPPET
 from tests.app.sep.conftest import REDUCED_ACTIVATION
 
 _ORIGINAL_SEP_APP = main_module.sep_app
@@ -295,6 +299,41 @@ def test_sep_app_rebuilds_without_alerts_and_dipper(mocker):
         _reload_restoring_identity()
 
 
+def test_embedded_activation_list_serves_a_snippet_download(mocker, tmp_path):
+    """Serve an ATW-dispatched snippet download with the snippets app deactivated.
+
+    Both halves of the artifact surface are import-time decisions — the mount in
+    ``main`` and ``_BASE_DIRS`` in the route module — so both are rebuilt against
+    the embedded activation list before the request. A 404 here means the router
+    was not mounted; a 400 means the snippet type did not resolve.
+    """
+    original_apps = sep_settings.APPS
+    original_sep_app = main_module.sep_app
+    (tmp_path / "collect.sh").write_text("#!/bin/bash\necho hello")
+    token = crypto_timestamp_serializer.dumps(
+        {"type": ARTIFACT_TYPE_SNIPPET, "filename": "collect.sh", "md5": "abc123"},
+        salt=ARTIFACT_DOWNLOAD_SALT,
+    )
+
+    mocker.patch.object(sep_settings, "APPS", REDUCED_ACTIVATION)
+    get_app_registry.cache_clear()
+    try:
+        importlib.reload(artifacts_module)
+        importlib.reload(main_module)
+
+        with patch("app.sep.snippets.config.snippets_settings.SNIPPETS_DIR", tmp_path):
+            client = TestClient(main_module.sep_app, raise_server_exceptions=False)
+            response = client.get(f"/artifacts/download/{token}")
+
+        assert response.status_code == status.HTTP_200_OK
+    finally:
+        sep_settings.APPS = original_apps
+        get_app_registry.cache_clear()
+        importlib.reload(artifacts_module)
+        importlib.reload(main_module)
+        main_module.sep_app = original_sep_app
+
+
 async def _refresher_proxy_map(mocker) -> dict[SettingClassEnum, ProxyEntry]:
     """Return the proxy map ``sep_overrides_lifespan`` hands to the refresher.
 

tests/sidecar/test_embedded_settings.py

@@ -30,6 +30,8 @@
     collect_app_owned_settings_classes,
 )
 from app.sep.config import SEPSettings
+from app.sep.routes.artifacts import collect_base_dirs
+from app.sep.snippets.constants import ARTIFACT_TYPE_SNIPPET
 from app.tasks.config import TasksSettings
 from app.tasks.settings.routes import TASKS_ADMIN_SETTINGS_CLASSES
 from tests.sidecar.conftest import (
@@ -267,6 +269,20 @@ def test_activation_list_builds_an_app_registry():
     assert "snippets" not in activated
 
 
+@pytest.mark.usefixtures("embedded_profile_cwd")
+def test_activation_list_resolves_the_snippet_artifact_type(mocker):
+    """Assert the baked profile serves ATW-dispatched snippet downloads.
+
+    The profile activates atw and no artifact-declaring app, so the type has to
+    come from the static map rather than the registry; without it the signed URL
+    ATW emits is rejected as an invalid artifact type.
+    """
+    registry = build_app_registry(SEPSettings().APPS)
+    mocker.patch("app.sep.routes.artifacts.get_app_registry", return_value=registry)
+
+    assert ARTIFACT_TYPE_SNIPPET in collect_base_dirs()
+
+
 @pytest.mark.usefixtures("embedded_profile_cwd")
 def test_uvicorn_ports_match_the_healthcheck_probe():
     """Assert the profile follows the probe's hardcoded ports, which are contract."""