cloudquery · marianogappa · May 12, 2026 · May 12, 2026 · May 12, 2026 · May 12, 2026
@@ -31,6 +31,7 @@ var docFiles = []string{
 	"cloudquery_plugin.md",
 	"cloudquery_plugin_install.md",
 	"cloudquery_plugin_publish.md",
+	"cloudquery_plugin_spec-schema.md",
 	"cloudquery_switch.md",
 }
 

@@ -0,0 +1,168 @@
+package cmd
+
+import (
+	"errors"
+	"fmt"
+	"os"
+	"path/filepath"
+
+	cqapiauth "github.com/cloudquery/cloudquery-api-go/auth"
+	"github.com/cloudquery/cloudquery/cli/v6/internal/auth"
+	"github.com/cloudquery/cloudquery/cli/v6/internal/hub"
+	"github.com/cloudquery/plugin-pb-go/managedplugin"
+	"github.com/cloudquery/plugin-pb-go/pb/plugin/v3"
+	"github.com/rs/zerolog/log"
+	"github.com/spf13/cobra"
+)
+
+const (
+	pluginSpecSchemaShort = "Export a plugin's spec JSON schema."
+	pluginSpecSchemaLong  = `Export a plugin's spec JSON schema.
+
+Only plugins published to the CloudQuery hub are supported. registry: local,
+registry: grpc, and registry: docker plugins are not exportable because they
+have no stable (path, version) identity to anchor the generated filename.
+For those registries the plugin binary is already accessible locally, so
+'cloudquery validate-config' can validate them in-place without --schemas-dir.
+
+Without --schemas-dir the schema is printed to stdout. With --schemas-dir the
+schema is written to <dir>/<plugin-name>@<version>.json, which is the
+filename format expected by ` + "`cloudquery validate-config --schemas-dir`" + `.
+Including the version in the filename ensures validation always runs against
+the schema matching the plugin version in the config.`
+	pluginSpecSchemaExample = `
+# Print schema to stdout
+cloudquery plugin spec-schema cloudquery/source/aws@v33.0.0
+
+# Write to ./schemas/aws@v33.0.0.json
+cloudquery plugin spec-schema cloudquery/source/aws@v33.0.0 -D ./schemas`
+)
+
+func newCmdPluginSpecSchema() *cobra.Command {
+	cmd := &cobra.Command{
+		Use:     "spec-schema <team_name>/<plugin_kind>/<plugin_name>@<version>",
+		Short:   pluginSpecSchemaShort,
+		Long:    pluginSpecSchemaLong,
+		Example: pluginSpecSchemaExample,
+		Args:    cobra.ExactArgs(1),
+		RunE:    runPluginSpecSchema,
+	}
+	cmd.Flags().StringP("schemas-dir", "D", "", "Write schema to <dir>/<plugin-name>@<version>.json. If omitted, the schema is printed to stdout.")
+	return cmd
+}
+
+func runPluginSpecSchema(cmd *cobra.Command, args []string) error {
+	schemasDir, err := cmd.Flags().GetString("schemas-dir")
+	if err != nil {
+		return err
+	}
+	cqDir, err := cmd.Flags().GetString("cq-dir")
+	if err != nil {
+		return err
+	}
+
+	ref, err := hub.ParseHubPluginRef(args[0])
+	if err != nil {
+		return err
+	}
+
+	pluginType, err := pluginTypeFromKind(ref.Kind)
+	if err != nil {
+		return err
+	}
+
+	ctx := cmd.Context()
+
+	// Only registry: cloudquery is supported. The hub-ref input format already
+	// constrains us to this registry, but the hardcoded value below is the
+	// single source of truth — do not add a flag that lets callers select
+	// local / grpc / docker without first defining how the exported filename
+	// (which is keyed on a stable name+version) should be derived for those.
+	pluginCfg := managedplugin.Config{
+		Name:     ref.Name,
+		Version:  ref.Version,
+		Path:     fmt.Sprintf("%s/%s", ref.TeamName, ref.Name),
+		Registry: managedplugin.RegistryCloudQuery,
+	}
+
+	// CloudQuery-registry plugins always need an auth token.
+	tc := cqapiauth.NewTokenClient()
+	authToken, err := tc.GetToken()
+	if err != nil {
+		return fmt.Errorf("failed to get auth token: %w", err)
+	}
+	teamName, err := auth.GetTeamForToken(ctx, authToken)
+	if err != nil {
+		return fmt.Errorf("failed to get team name: %w", err)
+	}
+
+	opts := []managedplugin.Option{
+		managedplugin.WithLogger(log.Logger),
+		managedplugin.WithAuthToken(authToken.Value),
+		managedplugin.WithTeamName(teamName),
+	}
+	if logConsole {
+		opts = append(opts, managedplugin.WithNoProgress())
+	}
+	if cqDir != "" {
+		opts = append(opts, managedplugin.WithDirectory(cqDir))
+	}
+	if disableSentry {
+		opts = append(opts, managedplugin.WithNoSentry())
+	}
+
+	clients, err := managedplugin.NewClients(ctx, pluginType, []managedplugin.Config{pluginCfg}, opts...)
+	if err != nil {
+		return enrichClientError(clients, []bool{false}, err)
+	}
+	defer func() {
+		if err := clients.Terminate(); err != nil {
+			fmt.Println(err)
+		}
+	}()
+	if len(clients) == 0 {
+		return errors.New("plugin client not initialized")
+	}
+
+	pluginClient := plugin.NewPluginClient(clients[0].Conn)
+	jsonSchema, err := getSpecSchemaFromPlugin(ctx, pluginClient)
+	if err != nil {
+		return fmt.Errorf("failed to fetch spec schema: %w", err)
+	}
+	if len(jsonSchema) == 0 {
+		return fmt.Errorf("plugin %s did not return a spec schema", ref.String())
+	}
+
+	return writeSchemaOutput(jsonSchema, ref.Name, ref.Version, schemasDir)
+}
+
+func pluginTypeFromKind(kind string) (managedplugin.PluginType, error) {
+	switch kind {
+	case "source":
+		return managedplugin.PluginSource, nil
+	case "destination":
+		return managedplugin.PluginDestination, nil
+	default:
+		return 0, fmt.Errorf("unsupported plugin kind %q (expected source or destination)", kind)
+	}
+}
+
+func writeSchemaOutput(jsonSchema, pluginName, pluginVersion, schemasDir string) error {
+	if schemasDir == "" {
+		_, err := fmt.Print(jsonSchema)
+		return err
+	}
+	if err := os.MkdirAll(schemasDir, 0o755); err != nil {
+		return err
+	}
+	return os.WriteFile(filepath.Join(schemasDir, schemaFileName(pluginName, pluginVersion)), []byte(jsonSchema), 0o644)
+}
+
+// schemaFileName returns the canonical filename for a plugin's schema under --schemas-dir.
+// Version is included whenever non-empty so consumers can pin validation to the right plugin version.
+func schemaFileName(pluginName, pluginVersion string) string {
+	if pluginVersion == "" {
+		return pluginName + ".json"
+	}
+	return pluginName + "@" + pluginVersion + ".json"
+}
@@ -0,0 +1,49 @@
+package cmd
+
+import (
+	"os"
+	"path"
+	"testing"
+
+	"github.com/cloudquery/plugin-pb-go/managedplugin"
+	"github.com/stretchr/testify/require"
+)
+
+func TestPluginTypeFromKind(t *testing.T) {
+	src, err := pluginTypeFromKind("source")
+	require.NoError(t, err)
+	require.Equal(t, managedplugin.PluginSource, src)
+
+	dst, err := pluginTypeFromKind("destination")
+	require.NoError(t, err)
+	require.Equal(t, managedplugin.PluginDestination, dst)
+
+	_, err = pluginTypeFromKind("transformer")
+	require.Error(t, err)
+}
+
+func TestSchemaFileName(t *testing.T) {
+	require.Equal(t, "aws@v33.0.0.json", schemaFileName("aws", "v33.0.0"))
+	require.Equal(t, "aws.json", schemaFileName("aws", ""))
+}
+
+func TestWriteSchemaOutput(t *testing.T) {
+	const schema = `{"type":"object"}`
+
+	t.Run("to schemas dir with versioned name", func(t *testing.T) {
+		dir := t.TempDir()
+		sub := path.Join(dir, "nested")
+		require.NoError(t, writeSchemaOutput(schema, "aws", "v33.0.0", sub))
+		got, err := os.ReadFile(path.Join(sub, "aws@v33.0.0.json"))
+		require.NoError(t, err)
+		require.Equal(t, schema, string(got))
+	})
+
+	t.Run("to schemas dir without version falls back to unversioned name", func(t *testing.T) {
+		dir := t.TempDir()
+		require.NoError(t, writeSchemaOutput(schema, "aws", "", dir))
+		got, err := os.ReadFile(path.Join(dir, "aws.json"))
+		require.NoError(t, err)
+		require.Equal(t, schema, string(got))
+	})
+}
@@ -190,6 +190,7 @@ func NewCmdRoot() *cobra.Command {
 	pluginCmd.AddCommand(
 		newCmdPluginInstall(false),
 		newCmdPluginPublish(),
+		newCmdPluginSpecSchema(),
 		pluginDocCmd,
 		pluginUIAssetsCmd,
 	)

@@ -115,22 +115,37 @@ func initPlugin(ctx context.Context, client plugin.PluginClient, spec map[string
 //  3. If the schema isn't empty but not valid, print the error message & skip the validation.
 //  4. Finally, return the validation result.
 func validatePluginSpec(ctx context.Context, client plugin.PluginClient, spec any) error {
+	jsonSchema, err := getSpecSchemaFromPlugin(ctx, client)
+	if err != nil {
+		return err
+	}
+	return validateSpecAgainstSchema(jsonSchema, spec)
+}
+
+func getSpecSchemaFromPlugin(ctx context.Context, client plugin.PluginClient) (string, error) {
 	schema, err := client.GetSpecSchema(ctx, &plugin.GetSpecSchema_Request{})
 	if err != nil {
 		st, ok := status.FromError(err)
 		if !ok {
 			// not a gRPC-compatible error
 			log.Err(err).Msg("failed to get spec schema")
-			return err
+			return "", err
 		}
 		if st.Code() != codes.Unimplemented {
 			// unimplemented is OK, treat as empty schema
 			log.Err(err).Msg("failed to get spec schema")
-			return err
+			return "", err
 		}
 	}
+	return schema.GetJsonSchema(), nil
+}
 
-	jsonSchema := schema.GetJsonSchema()
+// validateSpecAgainstSchema validates spec against a plugin-supplied JSON schema.
+// Intended for schemas obtained from a running plugin over gRPC: a malformed or empty
+// schema is treated as "skip validation" so a buggy plugin can't block sync/validate-config.
+// For user-supplied schema files use validateSpecAgainstSchemaStrict instead, which fails
+// loudly on a malformed schema rather than silently passing.
+func validateSpecAgainstSchema(jsonSchema string, spec any) error {
 	if len(jsonSchema) == 0 {
 		// This will also be true for Unimplemented response (schema = nil => schema.GetJsonSchema() = "")
 		log.Info().Msg("empty JSON schema for plugin spec, skipping validation")
@@ -146,6 +161,22 @@ func validatePluginSpec(ctx context.Context, client plugin.PluginClient, spec an
 	return sc.Validate(spec)
 }
 
+// validateSpecAgainstSchemaStrict validates spec against a JSON schema and treats parse
+// failures as errors. Use this when the schema is authoritative (e.g. a local file
+// passed via --schemas-dir) so a corrupt or empty schema cannot produce a false pass.
+func validateSpecAgainstSchemaStrict(jsonSchema string, spec any) error {
+	if len(jsonSchema) == 0 {
+		return errors.New("schema is empty")
+	}
+
+	sc, err := parseJSONSchema(jsonSchema)
+	if err != nil {
+		return fmt.Errorf("failed to parse JSON schema: %w", err)
+	}
+
+	return sc.Validate(spec)
+}
+
 func parseJSONSchema(jsonSchema string) (*jsonschema.Schema, error) {
 	c := jsonschema.NewCompiler()
 	c.DefaultDraft(jsonschema.Draft2020)

@@ -0,0 +1,40 @@
+package cmd
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestValidateSpecAgainstSchema_LenientForBuggyPlugins(t *testing.T) {
+	// Empty schema (e.g. plugin returned Unimplemented over gRPC) is treated as skip.
+	require.NoError(t, validateSpecAgainstSchema("", map[string]any{}))
+	// Unparseable schema is logged and skipped (lenient path used for plugin gRPC results).
+	require.NoError(t, validateSpecAgainstSchema(`{not-valid-json`, map[string]any{}))
+}
+
+func TestValidateSpecAgainstSchemaStrict_FailsOnBadSchemaAndBadSpec(t *testing.T) {
+	const goodSchema = `{
+		"$schema": "https://json-schema.org/draft/2020-12/schema",
+		"type": "object",
+		"properties": {"field": {"type": "string"}},
+		"required": ["field"],
+		"additionalProperties": false
+	}`
+
+	// Good spec passes.
+	require.NoError(t, validateSpecAgainstSchemaStrict(goodSchema, map[string]any{"field": "ok"}))
+
+	// Bad spec is rejected.
+	err := validateSpecAgainstSchemaStrict(goodSchema, map[string]any{"field": 42})
+	require.Error(t, err)
+
+	// Corrupt schema is rejected, NOT silently passed.
+	err = validateSpecAgainstSchemaStrict(`{not-valid-json`, map[string]any{})
+	require.Error(t, err)
+	require.Contains(t, err.Error(), "failed to parse JSON schema")
+
+	// Empty schema is rejected, NOT silently passed.
+	err = validateSpecAgainstSchemaStrict("", map[string]any{})
+	require.Error(t, err)
+}
@@ -0,0 +1,4 @@
+{
+  "$schema": "https://json-schema.org/draft/2020-12/schema",
+  "type": ["object", "null"]
+}
@@ -0,0 +1,8 @@
+{
+  "$schema": "https://json-schema.org/draft/2020-12/schema",
+  "type": ["object", "null"],
+  "additionalProperties": false,
+  "properties": {
+    "field": { "type": "string" }
+  }
+}
@@ -0,0 +1,15 @@
+kind: source
+spec:
+  name: src
+  path: ./nonexistent-src-binary
+  registry: local
+  destinations: [dst]
+  tables: ["*"]
+  spec:
+    bogus: field
+---
+kind: destination
+spec:
+  name: dst
+  path: ./nonexistent-dst-binary
+  registry: local
@@ -0,0 +1,13 @@
+kind: source
+spec:
+  name: src
+  path: ./nonexistent-src-binary
+  registry: local
+  destinations: [dst]
+  tables: ["*"]
+---
+kind: destination
+spec:
+  name: dst
+  path: ./nonexistent-dst-binary
+  registry: local