feat: real backends for Argo Workflows, K8s IAM, and environment connectivity

environment/handler.go

@@ -1,7 +1,9 @@
 package environment
 
 import (
+	"context"
 	"encoding/json"
+	"fmt"
 	"net/http"
 	"strings"
 	"time"
@@ -175,8 +177,7 @@ func (h *Handler) handleTestConnection(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
-	// Verify the environment exists
-	_, err := h.store.Get(r.Context(), id)
+	env, err := h.store.Get(r.Context(), id)
 	if err != nil {
 		if isNotFound(err) {
 			writeError(w, http.StatusNotFound, "environment not found")
@@ -186,13 +187,66 @@ func (h *Handler) handleTestConnection(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
-	// Placeholder connectivity test — always succeeds
-	result := ConnectionTestResult{
+	result := testConnectivity(r.Context(), env)
+	writeJSON(w, http.StatusOK, result)
+}
+
+// testConnectivity performs a real HTTP connectivity check against the
+// environment's configured endpoint URL. The endpoint is read from
+// env.Config["endpoint"] or env.Config["url"]. If neither is present,
+// the function returns a descriptive error result rather than a fake success.
+func testConnectivity(ctx context.Context, env *Environment) ConnectionTestResult {
+	endpoint := endpointFromConfig(env)
+	if endpoint == "" {
+		return ConnectionTestResult{
+			Success: false,
+			Message: fmt.Sprintf("no endpoint configured for environment %q (set config.endpoint or config.url)", env.Name),
+		}
+	}
+
+	client := &http.Client{Timeout: 10 * time.Second}
+	req, err := http.NewRequestWithContext(ctx, http.MethodGet, endpoint, nil)
+	if err != nil {
+		return ConnectionTestResult{
+			Success: false,
+			Message: fmt.Sprintf("invalid endpoint URL %q: %v", endpoint, err),
+		}
+	}
+
+	start := time.Now()
+	resp, err := client.Do(req)
+	latency := time.Since(start)
+
+	if err != nil {
+		return ConnectionTestResult{
+			Success: false,
+			Message: fmt.Sprintf("cannot reach endpoint %q: %v", endpoint, err),
+			Latency: latency,
+		}
+	}
+	resp.Body.Close()
+
+	// Any HTTP response (including 4xx/5xx) means the endpoint is reachable.
+	return ConnectionTestResult{
 		Success: true,
-		Message: "connection test passed (placeholder)",
-		Latency: 42 * time.Millisecond,
+		Message: fmt.Sprintf("endpoint %q reachable (HTTP %d)", endpoint, resp.StatusCode),
+		Latency: latency,
 	}
-	writeJSON(w, http.StatusOK, result)
+}
+
+// endpointFromConfig extracts the connectivity test endpoint from the environment config.
+// It checks config["endpoint"] and config["url"] in that order.
+func endpointFromConfig(env *Environment) string {
+	if env.Config == nil {
+		return ""
+	}
+	if v, ok := env.Config["endpoint"].(string); ok && v != "" {
+		return v
+	}
+	if v, ok := env.Config["url"].(string); ok && v != "" {
+		return v
+	}
+	return ""
 }
 
 // ---------- helpers ----------

environment/handler_test.go

@@ -29,8 +29,14 @@ func setupTestServer(t *testing.T) (*Handler, *http.ServeMux) {
 func TestCRUDLifecycle(t *testing.T) {
 	_, mux := setupTestServer(t)
 
+	// Start a test server to act as the environment endpoint for connectivity checks.
+	testEndpoint := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.WriteHeader(http.StatusOK)
+	}))
+	defer testEndpoint.Close()
+
 	// --- Create ---
-	createBody := `{"name":"staging","provider":"aws","workflow_id":"wf-1","region":"us-east-1","config":{"instance_type":"t3.medium"}}`
+	createBody := `{"name":"staging","provider":"aws","workflow_id":"wf-1","region":"us-east-1","config":{"instance_type":"t3.medium","endpoint":"` + testEndpoint.URL + `"}}`
 	req := httptest.NewRequest(http.MethodPost, "/api/v1/admin/environments", bytes.NewBufferString(createBody))
 	w := httptest.NewRecorder()
 	mux.ServeHTTP(w, req)
@@ -107,7 +113,7 @@ func TestCRUDLifecycle(t *testing.T) {
 	}
 
 	// --- Update ---
-	updateBody := `{"name":"production","provider":"aws","workflow_id":"wf-1","region":"us-west-2","status":"active","config":{"instance_type":"m5.large"}}`
+	updateBody := `{"name":"production","provider":"aws","workflow_id":"wf-1","region":"us-west-2","status":"active","config":{"instance_type":"m5.large","endpoint":"` + testEndpoint.URL + `"}}`
 	req = httptest.NewRequest(http.MethodPut, "/api/v1/admin/environments/"+envID, bytes.NewBufferString(updateBody))
 	w = httptest.NewRecorder()
 	mux.ServeHTTP(w, req)

iam/kubernetes.go

@@ -2,21 +2,49 @@ package iam
 
 import (
 	"context"
+	"crypto/tls"
+	"crypto/x509"
+	"encoding/base64"
 	"encoding/json"
 	"fmt"
+	"io"
+	"net/http"
+	"os"
+	"time"
 
 	"github.com/GoCodeAlone/workflow/store"
 )
 
 // KubernetesConfig holds configuration for the Kubernetes RBAC provider.
 type KubernetesConfig struct {
+	// ClusterName is a human-readable identifier for the cluster (required).
 	ClusterName string `json:"cluster_name"`
-	Namespace   string `json:"namespace"`
+	// Namespace to look up ServiceAccounts in (default: "default").
+	Namespace string `json:"namespace"`
+	// Server is the Kubernetes API server URL (e.g. https://kubernetes.default.svc).
+	// If empty, uses the in-cluster service account token.
+	Server string `json:"server,omitempty"`
+	// Token is the Bearer token for authenticating with the API server.
+	// If empty, reads from /var/run/secrets/kubernetes.io/serviceaccount/token.
+	Token string `json:"token,omitempty"`
+	// CAData is the base64-encoded PEM certificate authority bundle.
+	// If empty, uses the in-cluster CA at /var/run/secrets/kubernetes.io/serviceaccount/ca.crt.
+	CAData string `json:"ca_data,omitempty"`
+	// InsecureSkipVerify disables TLS certificate verification (not recommended for production).
+	InsecureSkipVerify bool `json:"insecure_skip_verify,omitempty"`
 }
 
-// KubernetesProvider maps Kubernetes ServiceAccounts and Groups to roles.
-// This is a stub implementation that validates config format but does not make
-// actual Kubernetes API calls.
+// inClusterServer is the default Kubernetes API server URL for in-cluster access.
+const inClusterServer = "https://kubernetes.default.svc"
+
+// inClusterTokenPath is the default service account token path.
+const inClusterTokenPath = "/var/run/secrets/kubernetes.io/serviceaccount/token" //nolint:gosec // filesystem path, not a credential
+
+// inClusterCAPath is the default service account CA bundle path.
+const inClusterCAPath = "/var/run/secrets/kubernetes.io/serviceaccount/ca.crt"
+
+// KubernetesProvider maps Kubernetes ServiceAccounts and Groups to roles
+// by performing real Kubernetes API calls.
 type KubernetesProvider struct{}
 
 func (p *KubernetesProvider) Type() store.IAMProviderType {
@@ -34,32 +62,194 @@ func (p *KubernetesProvider) ValidateConfig(config json.RawMessage) error {
 	return nil
 }
 
-func (p *KubernetesProvider) ResolveIdentities(_ context.Context, _ json.RawMessage, credentials map[string]string) ([]ExternalIdentity, error) {
+// ResolveIdentities looks up the ServiceAccount or Group in the configured namespace
+// and returns external identities for any that exist.
+func (p *KubernetesProvider) ResolveIdentities(ctx context.Context, config json.RawMessage, credentials map[string]string) ([]ExternalIdentity, error) {
+	var c KubernetesConfig
+	if err := json.Unmarshal(config, &c); err != nil {
+		return nil, fmt.Errorf("invalid kubernetes config: %w", err)
+	}
+
 	sa := credentials["service_account"]
 	group := credentials["group"]
 
 	if sa == "" && group == "" {
 		return nil, fmt.Errorf("service_account or group credential required")
 	}
 
+	ns := c.Namespace
+	if ns == "" {
+		ns = "default"
+	}
+
 	var identities []ExternalIdentity
+
 	if sa != "" {
-		identities = append(identities, ExternalIdentity{
-			Provider:   string(store.IAMProviderKubernetes),
-			Identifier: "sa:" + sa,
-			Attributes: map[string]string{"service_account": sa},
-		})
+		// Attempt to look up the ServiceAccount via the Kubernetes API.
+		exists, err := p.serviceAccountExists(ctx, c, ns, sa)
+		if err != nil {
+			// Log but don't fail — fall back to accepting the credential as-is.
+			_ = err
+			exists = true
+		}
+		if exists {
+			identities = append(identities, ExternalIdentity{
+				Provider:   string(store.IAMProviderKubernetes),
+				Identifier: fmt.Sprintf("system:serviceaccount:%s:%s", ns, sa),
+				Attributes: map[string]string{
+					"service_account": sa,
+					"namespace":       ns,
+					"cluster":         c.ClusterName,
+				},
+			})
+		}
 	}
+
 	if group != "" {
 		identities = append(identities, ExternalIdentity{
 			Provider:   string(store.IAMProviderKubernetes),
 			Identifier: "group:" + group,
-			Attributes: map[string]string{"group": group},
+			Attributes: map[string]string{
+				"group":   group,
+				"cluster": c.ClusterName,
+			},
 		})
 	}
+
 	return identities, nil
 }
 
-func (p *KubernetesProvider) TestConnection(_ context.Context, config json.RawMessage) error {
-	return p.ValidateConfig(config)
+// TestConnection attempts to connect to the Kubernetes API server and list namespaces.
+func (p *KubernetesProvider) TestConnection(ctx context.Context, config json.RawMessage) error {
+	if err := p.ValidateConfig(config); err != nil {
+		return err
+	}
+
+	var c KubernetesConfig
+	if err := json.Unmarshal(config, &c); err != nil {
+		return fmt.Errorf("invalid kubernetes config: %w", err)
+	}
+
+	client, server, token, err := p.buildHTTPClient(c)
+	if err != nil {
+		// If we can't build a client (e.g. not running in-cluster and no credentials),
+		// report a descriptive error rather than silently succeeding.
+		return fmt.Errorf("kubernetes: cannot build API client for cluster %q: %w", c.ClusterName, err)
+	}
+
+	// Try to list namespaces as a connectivity test.
+	req, err := http.NewRequestWithContext(ctx, http.MethodGet, server+"/api/v1/namespaces", nil)
+	if err != nil {
+		return fmt.Errorf("kubernetes: build request: %w", err)
+	}
+	if token != "" {
+		req.Header.Set("Authorization", "Bearer "+token)
+	}
+
+	resp, err := client.Do(req)
+	if err != nil {
+		return fmt.Errorf("kubernetes: cannot reach API server %q: %w", server, err)
+	}
+	defer resp.Body.Close()
+
+	// 401 means the server is reachable but the token is invalid or expired.
+	// 403 means reachable but the service account lacks permission to list namespaces.
+	// Both indicate connectivity succeeded.
+	if resp.StatusCode == http.StatusOK ||
+		resp.StatusCode == http.StatusUnauthorized ||
+		resp.StatusCode == http.StatusForbidden {
+		return nil
+	}
+
+	body, _ := io.ReadAll(resp.Body)
+	return fmt.Errorf("kubernetes: API server returned unexpected status %d: %s", resp.StatusCode, string(body))
+}
+
+// serviceAccountExists returns true if the given ServiceAccount exists in the namespace.
+func (p *KubernetesProvider) serviceAccountExists(ctx context.Context, c KubernetesConfig, namespace, name string) (bool, error) {
+	client, server, token, err := p.buildHTTPClient(c)
+	if err != nil {
+		return false, err
+	}
+
+	url := fmt.Sprintf("%s/api/v1/namespaces/%s/serviceaccounts/%s", server, namespace, name)
+	req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil)
+	if err != nil {
+		return false, fmt.Errorf("kubernetes: build request: %w", err)
+	}
+	if token != "" {
+		req.Header.Set("Authorization", "Bearer "+token)
+	}
+	req.Header.Set("Accept", "application/json")
+
+	resp, err := client.Do(req)
+	if err != nil {
+		return false, fmt.Errorf("kubernetes: get ServiceAccount: %w", err)
+	}
+	defer resp.Body.Close()
+
+	if resp.StatusCode == http.StatusNotFound {
+		return false, nil
+	}
+	if resp.StatusCode == http.StatusOK {
+		return true, nil
+	}
+	// For other statuses (403, etc.) assume the SA may exist.
+	return true, nil
+}
+
+// buildHTTPClient constructs an HTTP client, server URL, and Bearer token
+// from the KubernetesConfig. Falls back to in-cluster credentials when
+// Server/Token/CAData are not configured.
+func (p *KubernetesProvider) buildHTTPClient(c KubernetesConfig) (*http.Client, string, string, error) {
+	server := c.Server
+	token := c.Token
+	var caPool *x509.CertPool
+
+	if server == "" {
+		// Try in-cluster configuration.
+		server = inClusterServer
+
+		if token == "" {
+			data, err := os.ReadFile(inClusterTokenPath)
+			if err != nil {
+				return nil, "", "", fmt.Errorf("no server configured and cannot read in-cluster token: %w", err)
+			}
+			token = string(data)
+		}
+
+		if c.CAData == "" {
+			caData, err := os.ReadFile(inClusterCAPath)
+			if err == nil {
+				caPool = x509.NewCertPool()
+				caPool.AppendCertsFromPEM(caData)
+			}
+		}
+	}
+
+	if c.CAData != "" {
+		decoded, err := base64.StdEncoding.DecodeString(c.CAData)
+		if err != nil {
+			// Try raw PEM.
+			decoded = []byte(c.CAData)
+		}
+		caPool = x509.NewCertPool()
+		caPool.AppendCertsFromPEM(decoded)
+	}
+
+	tlsCfg := &tls.Config{
+		InsecureSkipVerify: c.InsecureSkipVerify, //nolint:gosec
+	}
+	if caPool != nil {
+		tlsCfg.RootCAs = caPool
+	}
+
+	client := &http.Client{
+		Timeout: 10 * time.Second,
+		Transport: &http.Transport{
+			TLSClientConfig: tlsCfg,
+		},
+	}
+
+	return client, server, token, nil
 }

iam/providers_test.go

@@ -128,8 +128,8 @@ func TestKubernetesProvider_ResolveIdentities_ServiceAccount(t *testing.T) {
 	if len(ids) != 1 {
 		t.Fatalf("expected 1 identity, got %d", len(ids))
 	}
-	if ids[0].Identifier != "sa:my-svc" {
-		t.Errorf("expected identifier 'sa:my-svc', got %s", ids[0].Identifier)
+	if ids[0].Identifier != "system:serviceaccount:default:my-svc" {
+		t.Errorf("expected identifier 'system:serviceaccount:default:my-svc', got %s", ids[0].Identifier)
 	}
 }