Validate image.source: registry requires a tag; warn on tag: latest

README.md

@@ -719,6 +719,8 @@ Common errors from `cds validate`, `cds plan`, and `cds render`, and how to fix
 | `[E041] ... Contract ref "x.y" points to unknown module "x"` | A `consumes` binding's `contractRef` refers to a module ID that isn't defined in the profile. | Check `spec.modules` for the correct module `id`, and confirm the contract ref follows `<module-id>.<contract-name>`. |
 | `[E041] ... but it does not provide "<contract-name>"` | The referenced module exists, but its `spec.provides` list doesn't expose that contract name. | Check the producing module's `module.yaml` for the contracts it actually provides, and fix the consumer's `contractRef` to match. |
 | `[E042] ... Contract kind mismatch` | The consumer expects one contract kind (e.g. `sql-database`) but the producer exposes a different kind. | Point the binding at a module that provides the expected contract kind, or update the consumer's expected kind if the mismatch is intentional. |
+| `[E103] ... config.image.tag is required ... when config.image.source is "registry"` | A module instance sets `config.image.source: registry` (e.g. Dagster) without also setting `config.image.tag`, so there's no published version to pull. | Set `config.image.tag` to a version published by `publish-images.yml` (optionally variant-prefixed, e.g. `hardened-1.8.0`), or switch back to `config.image.source: build`. |
+| `[W097] ... config.image.tag is "latest" with config.image.source "registry"` | `config.image.tag: latest` under `source: registry` still validates, but drifts silently between deploys instead of pinning a reproducible version. | Pin an explicit tag from `publish-images.yml`'s output (or `tests/fixtures/signed-images.json`) instead of `latest`. |
 
 All diagnostics print with their error code and YAML path (e.g. `spec.modules[1].config`), so search the profile file for that path to find the exact line to fix.
 

cli/validator.py

@@ -65,6 +65,7 @@ def validate_loaded_profile(profile: dict[str, Any], profile_file: Path) -> list
     diagnostics.extend(validate_dependencies(module_instances))
     diagnostics.extend(validate_secret_refs(profile, module_instances))
     diagnostics.extend(validate_contract_bindings(module_instances))
+    diagnostics.extend(validate_image_source_config(module_instances))
     diagnostics.extend(validate_outputs(profile, module_instances))
     diagnostics.extend(validate_observability_config(profile, module_instances))
 
@@ -409,6 +410,55 @@ def validate_contract_bindings(module_instances: list[dict[str, Any]]) -> list[D
     return diagnostics
 
 
+def validate_image_source_config(module_instances: list[dict[str, Any]]) -> list[Diagnostic]:
+    """
+    Modules that support pulling a pre-built image (`config.image.source:
+    registry`, e.g. modules/orchestration/dagster) need `config.image.tag`
+    set to something pullable; the module's own configSchema can't express
+    "tag is required only when source is registry" as a plain JSON Schema
+    constraint, so that cross-field rule is enforced here instead. A
+    `tag: "latest"` is accepted but discouraged, since it defeats the
+    reproducibility that `source: registry` is meant to buy over `latest`
+    silently drifting between deploys.
+    """
+    diagnostics: list[Diagnostic] = []
+
+    for inst in module_instances:
+        image_config = inst["config"].get("image")
+        if not isinstance(image_config, dict) or image_config.get("source") != "registry":
+            continue
+
+        tag = image_config.get("tag")
+        if not isinstance(tag, str) or not tag:
+            diagnostics.append(
+                Diagnostic(
+                    level="error",
+                    code="E103",
+                    message=(
+                        'config.image.tag is required and must be a non-empty string '
+                        'when config.image.source is "registry".'
+                    ),
+                    path=f"spec.modules[{inst['index']}].config.image.tag",
+                )
+            )
+            continue
+
+        if tag == "latest":
+            diagnostics.append(
+                Diagnostic(
+                    level="warning",
+                    code="W097",
+                    message=(
+                        'config.image.tag is "latest" with config.image.source "registry". '
+                        "Pin an explicit published version instead for reproducible deploys."
+                    ),
+                    path=f"spec.modules[{inst['index']}].config.image.tag",
+                )
+            )
+
+    return diagnostics
+
+
 def validate_outputs(profile: dict[str, Any], module_instances: list[dict[str, Any]]) -> list[Diagnostic]:
     diagnostics: list[Diagnostic] = []
 

tests/test_validator.py

@@ -8,6 +8,7 @@
 from cli.validator import (
     validate_contract_document,
     validate_contract_file,
+    validate_image_source_config,
     validate_observability_config,
     validate_profile,
 )
@@ -259,6 +260,103 @@ def test_sink_contract_ref_resolves_when_kind_matches(self):
         self.assertEqual(validate_observability_config(profile, module_instances), [])
 
 
+class ImageSourceConfigValidationTest(unittest.TestCase):
+    def _instance(self, config, index=0):
+        return {"index": index, "id": "under-test", "config": config}
+
+    def test_source_build_default_is_unaffected(self):
+        instances = [self._instance({"image": {"source": "build"}})]
+        self.assertEqual(validate_image_source_config(instances), [])
+
+    def test_absent_image_config_is_unaffected(self):
+        instances = [self._instance({})]
+        self.assertEqual(validate_image_source_config(instances), [])
+
+    def test_registry_source_without_tag_is_rejected(self):
+        instances = [self._instance({"image": {"source": "registry"}})]
+        diagnostics = validate_image_source_config(instances)
+        self.assertEqual([d.code for d in diagnostics], ["E103"])
+        self.assertEqual(diagnostics[0].path, "spec.modules[0].config.image.tag")
+
+    def test_registry_source_with_empty_tag_is_rejected(self):
+        instances = [self._instance({"image": {"source": "registry", "tag": ""}})]
+        diagnostics = validate_image_source_config(instances)
+        self.assertEqual([d.code for d in diagnostics], ["E103"])
+
+    def test_registry_source_with_pinned_tag_is_valid(self):
+        instances = [self._instance({"image": {"source": "registry", "tag": "1.8.0"}})]
+        self.assertEqual(validate_image_source_config(instances), [])
+
+    def test_registry_source_with_latest_tag_is_a_warning_not_an_error(self):
+        instances = [self._instance({"image": {"source": "registry", "tag": "latest"}})]
+        diagnostics = validate_image_source_config(instances)
+        self.assertEqual([d.code for d in diagnostics], ["W097"])
+        self.assertEqual(diagnostics[0].level, "warning")
+        self.assertEqual(diagnostics[0].path, "spec.modules[0].config.image.tag")
+
+    def test_full_profile_with_latest_tag_still_validates(self):
+        with tempfile.TemporaryDirectory() as tmpdir:
+            root = Path(tmpdir)
+            profile_dir = root / "profiles" / "local"
+            modules_root = root / "modules"
+            module_dir = modules_root / "orchestration" / "dagster"
+            profile_dir.mkdir(parents=True)
+            module_dir.mkdir(parents=True)
+
+            (module_dir / "module.yaml").write_text(
+                yaml.safe_dump(
+                    {
+                        "apiVersion": "cds/v1alpha1",
+                        "kind": "Module",
+                        "metadata": {"name": "dagster", "category": "orchestration", "version": "0.1.0"},
+                        "spec": {
+                            "runtime": {
+                                "type": "container",
+                                "service": {
+                                    "name": "dagster",
+                                    "ports": [{"name": "http", "containerPort": 3000, "protocol": "TCP"}],
+                                },
+                            },
+                            "configSchema": {
+                                "type": "object",
+                                "additionalProperties": True,
+                            },
+                            "implementation": {"kind": "docker-compose", "compose": {"services": {}}},
+                        },
+                    }
+                ),
+                encoding="utf-8",
+            )
+
+            profile = {
+                "apiVersion": "cds/v1alpha1",
+                "kind": "Profile",
+                "metadata": {"name": "local-test", "environment": "local"},
+                "spec": {
+                    "runtime": {"type": "docker-compose"},
+                    "modules": [
+                        {
+                            "id": "dagster",
+                            "source": "orchestration/dagster",
+                            "version": "0.1.0",
+                            "enabled": True,
+                            "config": {"image": {"source": "registry", "tag": "latest"}},
+                        }
+                    ],
+                    "secrets": {"provider": {"type": "env"}, "values": {}},
+                },
+            }
+
+            profile_file = profile_dir / "profile.yaml"
+            profile_file.write_text(yaml.safe_dump(profile), encoding="utf-8")
+
+            with patch.dict("os.environ", {"CDS_MODULE_PATH": str(modules_root)}, clear=False):
+                diagnostics = validate_profile(str(profile_file))
+
+            self.assertEqual([d for d in diagnostics if d.level == "error"], [])
+            self.assertEqual([d.code for d in diagnostics if d.level == "warning"], ["W097"])
+
+
 class ContractSchemaValidationTest(unittest.TestCase):
     def test_all_repo_contract_files_are_schema_valid(self):
         contract_files = sorted(_CONTRACTS_ROOT.glob("*.yaml"))