Implement CLI API Endpoints for CAPI Clustering (#498)

* Implement API/CLI for CAPI clustering

---------

Co-authored-by: Benjamin Schimke <benjamin.schimke@canonical.com>

src/k8s/api/v1/capi_config.go

@@ -0,0 +1,6 @@
+package v1
+
+// SetClusterAPIAuthTokenRequest is used to request to set the auth token for ClusterAPI.
+type SetClusterAPIAuthTokenRequest struct {
+	Token string `json:"token"`
+}

src/k8s/cmd/k8s/k8s.go

@@ -95,6 +95,7 @@ func NewRootCmd(env cmdutil.ExecutionEnvironment) *cobra.Command {
 		xPrintShimPidsCmd,
 		newXSnapdConfigCmd(env),
 		newXWaitForCmd(env),
+		newXCAPICmd(env),
 		newListImagesCmd(env),
 	)
 

src/k8s/cmd/k8s/k8s_x_capi.go

@@ -0,0 +1,47 @@
+package k8s
+
+import (
+	apiv1 "github.com/canonical/k8s/api/v1"
+	cmdutil "github.com/canonical/k8s/cmd/util"
+	"github.com/spf13/cobra"
+)
+
+func newXCAPICmd(env cmdutil.ExecutionEnvironment) *cobra.Command {
+	setAuthTokenCmd := &cobra.Command{
+		Use:   "set-auth-token <token>",
+		Short: "Set the auth token for the CAPI provider",
+		Args:  cmdutil.ExactArgs(env, 1),
+		Run: func(cmd *cobra.Command, args []string) {
+			token := args[0]
+			if token == "" {
+				cmd.PrintErrf("Error: The token must be provided.\n")
+				env.Exit(1)
+				return
+			}
+
+			client, err := env.Client(cmd.Context())
+			if err != nil {
+				cmd.PrintErrf("Error: Failed to create a k8sd client. Make sure that the k8sd service is running.\n\nThe error was: %v\n", err)
+				env.Exit(1)
+				return
+			}
+
+			err = client.SetClusterAPIAuthToken(cmd.Context(), apiv1.SetClusterAPIAuthTokenRequest{Token: token})
+			if err != nil {
+				cmd.PrintErrf("Error: Failed to set the CAPI auth token.\n\nThe error was: %v\n", err)
+				env.Exit(1)
+				return
+			}
+		},
+	}
+
+	cmd := &cobra.Command{
+		Use:    "x-capi",
+		Short:  "Manage the CAPI integration",
+		Hidden: true,
+	}
+
+	cmd.AddCommand(setAuthTokenCmd)
+
+	return cmd
+}

src/k8s/pkg/k8s/client/capi_auth.go

@@ -0,0 +1,17 @@
+package client
+
+import (
+	"context"
+	"fmt"
+
+	apiv1 "github.com/canonical/k8s/api/v1"
+	"github.com/canonical/lxd/shared/api"
+)
+
+// SetClusterAPIAuthToken calls "POST 1.0/x/capi/set-auth-token".
+func (c *k8sdClient) SetClusterAPIAuthToken(ctx context.Context, request apiv1.SetClusterAPIAuthTokenRequest) error {
+	if err := c.mc.Query(ctx, "POST", api.NewURL().Path("x", "capi", "set-auth-token"), request, nil); err != nil {
+		return fmt.Errorf("failed to POST /x/capi/set-auth-token: %w", err)
+	}
+	return nil
+}

src/k8s/pkg/k8s/client/interface.go

@@ -34,6 +34,8 @@ type Client interface {
 	UpdateClusterConfig(ctx context.Context, request apiv1.UpdateClusterConfigRequest) error
 	// GetClusterConfig retrieves configuration of the cluster.
 	GetClusterConfig(ctx context.Context, request apiv1.GetClusterConfigRequest) (apiv1.UserFacingClusterConfig, error)
+	// SetClusterAPIAuthToken sets the auth token for the CAPI provider.
+	SetClusterAPIAuthToken(ctx context.Context, request apiv1.SetClusterAPIAuthTokenRequest) error
 }
 
 var _ Client = &k8sdClient{}

src/k8s/pkg/k8s/client/mock/mock.go

@@ -47,8 +47,10 @@ type Client struct {
 		Config apiv1.UserFacingClusterConfig
 		Err    error
 	}
-	UpdateClusterConfigCalledWith apiv1.UpdateClusterConfigRequest
-	UpdateClusterConfigErr        error
+	UpdateClusterConfigCalledWith    apiv1.UpdateClusterConfigRequest
+	UpdateClusterConfigErr           error
+	SetClusterAPIAuthTokenCalledWith apiv1.SetClusterAPIAuthTokenRequest
+	SetClusterAPIAuthTokenErr        error
 }
 
 func (c *Client) Bootstrap(ctx context.Context, request apiv1.PostClusterBootstrapRequest) (apiv1.NodeStatus, error) {
@@ -113,4 +115,9 @@ func (c *Client) GetClusterConfig(ctx context.Context, request apiv1.GetClusterC
 	return c.GetClusterConfigReturn.Config, c.GetClusterConfigReturn.Err
 }
 
+func (c *Client) SetClusterAPIAuthToken(ctx context.Context, request apiv1.SetClusterAPIAuthTokenRequest) error {
+	c.SetClusterAPIAuthTokenCalledWith = request
+	return c.SetClusterAPIAuthTokenErr
+}
+
 var _ client.Client = &Client{}

src/k8s/pkg/k8sd/api/capi_access_handler.go

@@ -0,0 +1,38 @@
+package api
+
+import (
+	"context"
+	"database/sql"
+	"fmt"
+	"net/http"
+
+	"github.com/canonical/k8s/pkg/k8sd/database"
+	"github.com/canonical/lxd/lxd/response"
+	"github.com/canonical/microcluster/state"
+)
+
+func ValidateCAPIAuthTokenAccessHandler(tokenHeaderName string) func(s *state.State, r *http.Request) response.Response {
+	return func(s *state.State, r *http.Request) response.Response {
+		token := r.Header.Get(tokenHeaderName)
+		if token == "" {
+			return response.Unauthorized(fmt.Errorf("missing header %q", tokenHeaderName))
+		}
+
+		var tokenIsValid bool
+		if err := s.Database.Transaction(s.Context, func(ctx context.Context, tx *sql.Tx) error {
+			var err error
+			tokenIsValid, err = database.ValidateClusterAPIToken(ctx, tx, token)
+			if err != nil {
+				return fmt.Errorf("failed to check CAPI auth token: %w", err)
+			}
+			return nil
+		}); err != nil {
+			return response.InternalError(fmt.Errorf("check CAPI auth token database transaction failed: %w", err))
+		}
+		if !tokenIsValid {
+			return response.Unauthorized(fmt.Errorf("invalid token"))
+		}
+
+		return response.EmptySyncResponse
+	}
+}

src/k8s/pkg/k8sd/api/capi_auth.go

@@ -0,0 +1,29 @@
+package api
+
+import (
+	"context"
+	"database/sql"
+	"encoding/json"
+	"fmt"
+	"net/http"
+
+	apiv1 "github.com/canonical/k8s/api/v1"
+	"github.com/canonical/k8s/pkg/k8sd/database"
+	"github.com/canonical/lxd/lxd/response"
+	"github.com/canonical/microcluster/state"
+)
+
+func (e *Endpoints) postSetClusterAPIAuthToken(s *state.State, r *http.Request) response.Response {
+	request := apiv1.SetClusterAPIAuthTokenRequest{}
+	if err := json.NewDecoder(r.Body).Decode(&request); err != nil {
+		return response.BadRequest(fmt.Errorf("failed to parse request: %w", err))
+	}
+
+	if err := s.Database.Transaction(r.Context(), func(ctx context.Context, tx *sql.Tx) error {
+		return database.SetClusterAPIToken(ctx, tx, request.Token)
+	}); err != nil {
+		return response.InternalError(err)
+	}
+
+	return response.SyncResponse(true, nil)
+}

src/k8s/pkg/k8sd/api/endpoints.go

@@ -89,5 +89,16 @@ func (e *Endpoints) Endpoints() []rest.Endpoint {
 			Path: "kubernetes/auth/webhook",
 			Post: rest.EndpointAction{Handler: e.postKubernetesAuthWebhook, AllowUntrusted: true},
 		},
+		// ClusterAPI management endpoints.
+		{
+			Name: "ClusterAPI/GenerateJoinToken",
+			Path: "x/capi/generate-join-token",
+			Post: rest.EndpointAction{Handler: e.postClusterJoinTokens, AccessHandler: ValidateCAPIAuthTokenAccessHandler("capi-auth-token"), AllowUntrusted: true},
+		},
+		{
+			Name: "ClusterAPI/SetAuthToken",
+			Path: "x/capi/set-auth-token",
+			Post: rest.EndpointAction{Handler: e.postSetClusterAPIAuthToken},
+		},
 	}
 }

src/k8s/pkg/k8sd/database/capi_auth.go

@@ -0,0 +1,49 @@
+package database
+
+import (
+	"context"
+	"database/sql"
+	"fmt"
+
+	"github.com/canonical/microcluster/cluster"
+)
+
+var (
+	clusterAPIConfigsStmts = map[string]int{
+		"insert-capi-token": MustPrepareStatement("cluster-configs", "insert-capi-token.sql"),
+		"select-capi-token": MustPrepareStatement("cluster-configs", "select-capi-token.sql"),
+	}
+)
+
+// SetClusterAPIToken stores the ClusterAPI token in the cluster config.
+func SetClusterAPIToken(ctx context.Context, tx *sql.Tx, token string) error {
+	if token == "" {
+		return fmt.Errorf("token cannot be empty")
+	}
+
+	insertTxStmt, err := cluster.Stmt(tx, clusterAPIConfigsStmts["insert-capi-token"])
+	if err != nil {
+		return fmt.Errorf("failed to prepare insert statement: %w", err)
+	}
+	if _, err := insertTxStmt.ExecContext(ctx, token); err != nil {
+		return fmt.Errorf("insert ClusterAPI token query failed: %w", err)
+	}
+
+	return nil
+}
+
+// ValidateClusterAPIToken returns true if the specified token matches the stored ClusterAPI token.
+func ValidateClusterAPIToken(ctx context.Context, tx *sql.Tx, token string) (bool, error) {
+	selectTxStmt, err := cluster.Stmt(tx, clusterAPIConfigsStmts["select-capi-token"])
+	if err != nil {
+		return false, fmt.Errorf("failed to prepare select statement: %w", err)
+	}
+
+	var exists bool
+	err = selectTxStmt.QueryRowContext(ctx, token).Scan(&exists)
+	if err != nil {
+		return false, fmt.Errorf("failed to query ClusterAPI token: %w", err)
+	}
+
+	return exists, nil
+}

src/k8s/pkg/k8sd/database/capi_auth_test.go

@@ -0,0 +1,50 @@
+package database_test
+
+import (
+	"context"
+	"database/sql"
+	"testing"
+
+	"github.com/canonical/k8s/pkg/k8sd/database"
+	. "github.com/onsi/gomega"
+)
+
+func TestClusterAPIAuthTokens(t *testing.T) {
+	WithDB(t, func(ctx context.Context, db DB) {
+		var token string = "test-token"
+
+		t.Run("SetAuthToken", func(t *testing.T) {
+			g := NewWithT(t)
+			err := db.Transaction(ctx, func(ctx context.Context, tx *sql.Tx) error {
+				err := database.SetClusterAPIToken(ctx, tx, token)
+				g.Expect(err).To(BeNil())
+				return nil
+			})
+			g.Expect(err).To(BeNil())
+		})
+
+		t.Run("CheckAuthToken", func(t *testing.T) {
+			t.Run("ValidToken", func(t *testing.T) {
+				g := NewWithT(t)
+				err := db.Transaction(ctx, func(ctx context.Context, tx *sql.Tx) error {
+					valid, err := database.ValidateClusterAPIToken(ctx, tx, token)
+					g.Expect(err).To(BeNil())
+					g.Expect(valid).To(BeTrue())
+					return nil
+				})
+				g.Expect(err).To(BeNil())
+			})
+
+			t.Run("InvalidToken", func(t *testing.T) {
+				g := NewWithT(t)
+				err := db.Transaction(ctx, func(ctx context.Context, tx *sql.Tx) error {
+					valid, err := database.ValidateClusterAPIToken(ctx, tx, "invalid-token")
+					g.Expect(err).To(BeNil())
+					g.Expect(valid).To(BeFalse())
+					return nil
+				})
+				g.Expect(err).To(BeNil())
+			})
+		})
+	})
+}

src/k8s/pkg/k8sd/database/sql/queries/cluster-configs/insert-capi-token.sql

@@ -0,0 +1,7 @@
+INSERT INTO
+    cluster_configs(key, value)
+VALUES
+    ("token::capi", ?)
+ON CONFLICT(key) DO
+    UPDATE SET value = EXCLUDED.value;
+

src/k8s/pkg/k8sd/database/sql/queries/cluster-configs/select-capi-token.sql

@@ -0,0 +1,6 @@
+SELECT
+    EXISTS (
+        SELECT 1
+        FROM cluster_configs AS c
+        WHERE c.key = 'token::capi' AND c.value = ?
+    )

tests/integration/tests/test_smoke.py

@@ -1,6 +1,7 @@
 #
 # Copyright 2024 Canonical, Ltd.
 #
+import json
 import logging
 from typing import List
 
@@ -54,3 +55,40 @@ def test_smoke(instances: List[harness.Instance]):
             ["cat", f"/var/snap/k8s/common/args/{service}"], capture_output=True
         )
         assert value in args.stdout.decode()
+
+    LOG.info("Verify the functionality of the CAPI endpoints.")
+    instance.exec("k8s x-capi set-auth-token my-secret-token".split())
+
+    body = {
+        "name": "my-node",
+        "worker": False,
+    }
+
+    resp = instance.exec(
+        [
+            "curl",
+            "-XPOST",
+            "-H",
+            "Content-Type: application/json",
+            "-H",
+            "capi-auth-token: my-secret-token",
+            "--data",
+            json.dumps(body),
+            "--unix-socket",
+            "/var/snap/k8s/common/var/lib/k8sd/state/control.socket",
+            "http://localhost/1.0/x/capi/generate-join-token",
+        ],
+        capture_output=True,
+    )
+    response = json.loads(resp.stdout.decode())
+
+    assert (
+        response["error_code"] == 0
+    ), "Failed to generate join token using CAPI endpoints."
+    metadata = response.get("metadata")
+    assert (
+        metadata is not None
+    ), "Metadata not found in the generate-join-token response."
+    assert (
+        metadata.get("token") is not None
+    ), "Token not found in the generate-join-token response."