Add semantic validator for policy template datastream categories

code/go/internal/validator/semantic/validate_datastream_package_categories.go

@@ -0,0 +1,126 @@
+// Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+// or more contributor license agreements. Licensed under the Elastic License;
+// you may not use this file except in compliance with the Elastic License.
+
+package semantic
+
+import (
+	"fmt"
+	"io"
+	"io/fs"
+	"net/http"
+	"path"
+	"slices"
+	"sort"
+	"time"
+
+	"gopkg.in/yaml.v3"
+
+	"github.com/elastic/package-spec/v3/code/go/internal/fspath"
+	"github.com/elastic/package-spec/v3/code/go/pkg/specerrors"
+)
+
+const packageRegistryCategoriesURL = "https://raw.githubusercontent.com/elastic/package-registry/main/categories/categories.yml"
+
+type registryCategories struct {
+	Categories map[string]struct {
+		Title         string `yaml:"title"`
+		Subcategories map[string]struct {
+			Title string `yaml:"title"`
+		} `yaml:"subcategories"`
+	} `yaml:"categories"`
+}
+
+func fetchRegistryParentCategories() (map[string]struct{}, error) {
+	client := &http.Client{Timeout: 10 * time.Second}
+	resp, err := client.Get(packageRegistryCategoriesURL)
+	if err != nil {
+		return nil, fmt.Errorf("failed to fetch categories from package registry: %w", err)
+	}
+	defer resp.Body.Close()
+
+	body, err := io.ReadAll(resp.Body)
+	if err != nil {
+		return nil, fmt.Errorf("failed to read categories response: %w", err)
+	}
+
+	var rc registryCategories
+	if err := yaml.Unmarshal(body, &rc); err != nil {
+		return nil, fmt.Errorf("failed to parse categories YAML: %w", err)
+	}
+
+	parentCategories := make(map[string]struct{}, len(rc.Categories))
+	for id := range rc.Categories {
+		parentCategories[id] = struct{}{}
+	}
+	return parentCategories, nil
+}
+
+type packageManifestWithPackageCategories struct {
+	Type       string   `yaml:"type"`
+	Categories []string `yaml:"categories"`
+}
+
+// ValidateDatastreamPackageCategories validates that the package manifest
+// categories include all parent-level categories present in any data stream
+// manifest. Parent categories are determined by fetching the package registry
+// categories.yml. Data stream manifests without a categories field are skipped.
+func ValidateDatastreamPackageCategories(fsys fspath.FS) specerrors.ValidationErrors {
+	manifestPath := "manifest.yml"
+	data, err := fs.ReadFile(fsys, manifestPath)
+	if err != nil {
+		return specerrors.ValidationErrors{
+			specerrors.NewStructuredErrorf("file \"%s\" is invalid: %w", fsys.Path(manifestPath), errFailedToReadManifest)}
+	}
+
+	var manifest packageManifestWithPackageCategories
+	if err := yaml.Unmarshal(data, &manifest); err != nil {
+		return specerrors.ValidationErrors{
+			specerrors.NewStructuredErrorf("file \"%s\" is invalid: %w", fsys.Path(manifestPath), errFailedToParseManifest)}
+	}
+
+	if manifest.Type != packageTypeIntegration {
+		return nil
+	}
+
+	dsCategories, err := readDataStreamManifestCategories(fsys)
+	if err != nil {
+		return specerrors.ValidationErrors{
+			specerrors.NewStructuredErrorf("file \"%s\" is invalid: %w", fsys.Path(manifestPath), err)}
+	}
+
+	if len(dsCategories) == 0 {
+		return nil
+	}
+
+	parentCategories, err := fetchRegistryParentCategories()
+	if err != nil {
+		return specerrors.ValidationErrors{
+			specerrors.NewStructuredErrorf("file \"%s\" is invalid: failed to load registry categories: %w", fsys.Path(manifestPath), err)}
+	}
+
+	var errs specerrors.ValidationErrors
+	for dsName, dsCats := range dsCategories {
+		var missingCats []string
+		for _, dsCat := range dsCats {
+			if _, isParent := parentCategories[dsCat]; isParent && !slices.Contains(manifest.Categories, dsCat) {
+				missingCats = append(missingCats, dsCat)
+			}
+		}
+
+		if len(missingCats) > 0 {
+			sort.Strings(missingCats)
+			dsManifestPath := path.Join("data_stream", dsName, "manifest.yml")
+			errs = append(errs, specerrors.NewStructuredErrorf(
+				"file \"%s\" is invalid: package manifest categories %v are missing parent categories %v from data stream \"%s\" (defined in \"%s\")",
+				fsys.Path(manifestPath),
+				manifest.Categories,
+				missingCats,
+				dsName,
+				fsys.Path(dsManifestPath),
+			))
+		}
+	}
+
+	return errs
+}

code/go/internal/validator/semantic/validate_datastream_package_categories_test.go

@@ -0,0 +1,197 @@
+// Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+// or more contributor license agreements. Licensed under the Elastic License;
+// you may not use this file except in compliance with the Elastic License.
+
+package semantic
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/elastic/package-spec/v3/code/go/internal/fspath"
+)
+
+func TestValidateDatastreamPackageCategories(t *testing.T) {
+	cases := []struct {
+		title        string
+		setup        func(t *testing.T, dir string)
+		expectedErrs []string
+	}{
+		{
+			title: "package includes datastream parent category",
+			setup: func(t *testing.T, dir string) {
+				writeManifest(t, dir, `
+type: integration
+categories:
+  - security
+`)
+				writeDataStreamManifest(t, dir, "mylogs", `
+title: My Logs
+categories:
+  - security
+type: logs
+`)
+			},
+		},
+		{
+			title: "package missing datastream parent category",
+			setup: func(t *testing.T, dir string) {
+				writeManifest(t, dir, `
+type: integration
+categories:
+  - observability
+`)
+				writeDataStreamManifest(t, dir, "mylogs", `
+title: My Logs
+categories:
+  - security
+type: logs
+`)
+			},
+			expectedErrs: []string{
+				`package manifest categories [observability] are missing parent categories [security] from data stream "mylogs"`,
+			},
+		},
+		{
+			title: "datastream has subcategory only, no parent to enforce",
+			setup: func(t *testing.T, dir string) {
+				writeManifest(t, dir, `
+type: integration
+categories:
+  - observability
+`)
+				writeDataStreamManifest(t, dir, "mylogs", `
+title: My Logs
+categories:
+  - credential_management
+type: logs
+`)
+			},
+		},
+		{
+			title: "data stream manifest has no categories field, skipped",
+			setup: func(t *testing.T, dir string) {
+				writeManifest(t, dir, `
+type: integration
+categories:
+  - observability
+`)
+				writeDataStreamManifest(t, dir, "mylogs", `
+title: My Logs
+type: logs
+`)
+			},
+		},
+		{
+			title: "package has no categories but datastream does",
+			setup: func(t *testing.T, dir string) {
+				writeManifest(t, dir, `
+type: integration
+`)
+				writeDataStreamManifest(t, dir, "mylogs", `
+title: My Logs
+categories:
+  - security
+type: logs
+`)
+			},
+			expectedErrs: []string{
+				`package manifest categories [] are missing parent categories [security] from data stream "mylogs"`,
+			},
+		},
+		{
+			title: "package has subset, missing one datastream parent category",
+			setup: func(t *testing.T, dir string) {
+				writeManifest(t, dir, `
+type: integration
+categories:
+  - security
+`)
+				writeDataStreamManifest(t, dir, "mylogs", `
+title: My Logs
+categories:
+  - security
+  - observability
+type: logs
+`)
+			},
+			expectedErrs: []string{`are missing parent categories [observability] from data stream "mylogs"`},
+		},
+		{
+			title: "non-integration packages are skipped",
+			setup: func(t *testing.T, dir string) {
+				writeManifest(t, dir, `
+type: input
+categories:
+  - observability
+`)
+				writeDataStreamManifest(t, dir, "mylogs", `
+title: My Logs
+categories:
+  - security
+type: logs
+`)
+			},
+		},
+		{
+			title: "multiple datastreams, one triggers error",
+			setup: func(t *testing.T, dir string) {
+				writeManifest(t, dir, `
+type: integration
+categories:
+  - security
+`)
+				writeDataStreamManifest(t, dir, "auditlogs", `
+title: Audit Logs
+categories:
+  - security
+type: logs
+`)
+				writeDataStreamManifest(t, dir, "netlogs", `
+title: Net Logs
+categories:
+  - network
+type: logs
+`)
+			},
+			expectedErrs: []string{`are missing parent categories [network] from data stream "netlogs"`},
+		},
+		{
+			title: "datastream does not need all package categories",
+			setup: func(t *testing.T, dir string) {
+				writeManifest(t, dir, `
+type: integration
+categories:
+  - security
+  - observability
+`)
+				writeDataStreamManifest(t, dir, "mylogs", `
+title: My Logs
+categories:
+  - security
+type: logs
+`)
+			},
+		},
+	}
+
+	for _, tc := range cases {
+		t.Run(tc.title, func(t *testing.T) {
+			dir := t.TempDir()
+			tc.setup(t, dir)
+
+			errs := ValidateDatastreamPackageCategories(fspath.DirFS(dir))
+
+			if len(tc.expectedErrs) == 0 {
+				assert.Empty(t, errs)
+			} else {
+				require.Len(t, errs, 1)
+				for _, expected := range tc.expectedErrs {
+					assert.Contains(t, errs[0].Error(), expected)
+				}
+			}
+		})
+	}
+}