polaris/pkg/kube/resources.go

// Copyright 2022 FairwindsOps, Inc.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

package kube

import (
	"bytes"
	"context"
	"encoding/json"
	"fmt"
	"io"
	"os"
	"path/filepath"
	"regexp"
	"strings"
	"time"

	"github.com/fairwindsops/controller-utils/pkg/controller"
	conf "github.com/fairwindsops/polaris/pkg/config"

	"github.com/sirupsen/logrus"
	"github.com/thoas/go-funk"
	corev1 "k8s.io/api/core/v1"
	"k8s.io/apimachinery/pkg/api/meta"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/runtime/schema"
	k8sYaml "k8s.io/apimachinery/pkg/util/yaml"
	"k8s.io/client-go/dynamic"
	"k8s.io/client-go/kubernetes"
	_ "k8s.io/client-go/plugin/pkg/client/auth" // Required for other auth providers like GKE.
	"k8s.io/client-go/rest"
	"k8s.io/client-go/restmapper"
	"sigs.k8s.io/controller-runtime/pkg/client/config"
)

// ResourceProvider contains k8s resources to be audited
type ResourceProvider struct {
	ServerVersion string
	CreationTime  time.Time
	SourceName    string
	SourceType    string
	Nodes         []corev1.Node
	Namespaces    []corev1.Namespace
	Pods          []corev1.Pod
	Resources     resourceKindMap
}

type resourceKindMap map[string][]GenericResource

func (rkm resourceKindMap) addResource(r GenericResource) {
	gvk := r.Resource.GroupVersionKind()
	var key string
	if gvk.Group != "" {
		key = gvk.Group + "/" + gvk.Kind
	} else {
		key = gvk.Kind
	}
	rkm[key] = append(rkm[key], r)
}

func (rkm resourceKindMap) addResources(rs []GenericResource) {
	for _, r := range rs {
		rkm.addResource(r)
	}
}

func (rkm resourceKindMap) GetLength() int {
	total := 0
	for _, rs := range rkm {
		total += len(rs)
	}
	return total
}

func (rkm resourceKindMap) GetNumberOfControllers() int {
	total := 0
	for _, rs := range rkm {
		for _, r := range rs {
			if r.PodSpec != nil {
				total++
			}
		}
	}
	return total
}

var kindRewrites = map[string]string{
	"Ingress":                 "networking.k8s.io/Ingress",
	"PodDisruptionBudget":     "policy/PodDisruptionBudget",
	"HorizontalPodAutoscaler": "autoscaling/HorizontalPodAutoscaler",
}

// This is here for backward compatibility reasons
func maybeTransformKindIntoGroupKind(k string) string {
	if val, ok := kindRewrites[k]; ok {
		return val
	}
	return k
}

func parseGroupKind(gk string) schema.GroupKind {
	before, after, ok := strings.Cut(gk, "/")
	if !ok {
		return schema.GroupKind{Kind: gk}
	}

	group := before
	kind := after
	return schema.GroupKind{Group: group, Kind: kind}
}

func newResourceProvider(version, sourceType, sourceName string) ResourceProvider {
	return ResourceProvider{
		ServerVersion: version,
		SourceType:    sourceType,
		SourceName:    sourceName,
		CreationTime:  time.Now(),
		Nodes:         make([]corev1.Node, 0),
		Namespaces:    make([]corev1.Namespace, 0),
		Pods:          make([]corev1.Pod, 0),
		Resources:     make(map[string][]GenericResource),
	}
}

type k8sResource struct {
	Kind string `yaml:"kind"`
}

var podSpecFields = []string{"jobTemplate", "spec", "template"}

// CreateResourceProvider returns a new ResourceProvider object to interact with k8s resources
func CreateResourceProvider(ctx context.Context, directory, workload string, c conf.Configuration) (*ResourceProvider, error) {
	if workload != "" {
		return CreateResourceProviderFromResource(ctx, workload)
	}
	if directory != "" {
		return CreateResourceProviderFromPath(directory)
	}
	return CreateResourceProviderFromCluster(ctx, c)
}

// CreateResourceProviderFromResource creates a new ResourceProvider that just contains one workload
func CreateResourceProviderFromResource(ctx context.Context, workload string) (*ResourceProvider, error) {
	dynamicClient, restMapper, clientSet, _, err := GetKubeClient(ctx, "")
	if err != nil {
		return nil, err
	}
	serverVersion, err := clientSet.Discovery().ServerVersion()
	if err != nil {
		return nil, fmt.Errorf("Error fetching Cluster API version: %w", err)
	}
	resources := newResourceProvider(serverVersion.Major+"."+serverVersion.Minor, "Resource", workload)

	parts := strings.Split(workload, "/")
	if len(parts) != 4 {
		return nil, fmt.Errorf("Invalid workload identifier %s. Should be in format namespace/kind/version/name, e.g. nginx-ingress/Deployment.apps/v1/default-backend", workload)
	}
	namespace := parts[0]
	kind := parts[1]
	version := parts[2]
	name := parts[3]

	obj, err := GetObject(ctx, namespace, kind, version, name, dynamicClient, restMapper)
	if err != nil {
		return nil, fmt.Errorf("Could not find workload %s: %w", workload, err)
	}
	workloadObj, err := NewGenericResourceFromUnstructured(*obj, nil)
	if err != nil {
		return nil, fmt.Errorf("Could not parse workload %s: %w", workload, err)
	}
	resources.Resources.addResource(workloadObj)
	return &resources, nil
}

// CreateResourceProviderFromPath returns a new ResourceProvider using the YAML files in a directory
func CreateResourceProviderFromPath(directory string) (*ResourceProvider, error) {
	resources := newResourceProvider("unknown", "Path", directory)

	if directory == "-" {
		fi, err := os.Stdin.Stat()
		if err == nil && fi.Mode()&os.ModeNamedPipe == os.ModeNamedPipe {
			if err := resources.addResourcesFromReader(os.Stdin); err != nil {
				return nil, err
			}
			return &resources, nil
		}
	}

	visitFile := func(path string, f os.FileInfo, err error) error {
		if !strings.HasSuffix(path, ".yml") && !strings.HasSuffix(path, ".yaml") {
			return nil
		}
		contents, err := os.ReadFile(path)
		if err != nil {
			logrus.Errorf("Error reading file: %v", path)
			return err
		}
		err = resources.addResourcesFromYaml(string(contents))
		if err != nil {
			logrus.Warnf("skipping %s: cannot add resource from YAML: %v", path, err)
		}
		return nil
	}

	err := filepath.Walk(directory, visitFile)
	if err != nil {
		return nil, err
	}
	return &resources, nil
}

// CreateResourceProviderFromYaml returns a new ResourceProvider using the yaml
func CreateResourceProviderFromYaml(yamlContent string) (*ResourceProvider, error) {
	resources := newResourceProvider("unknown", "Content", "unknown")
	err := resources.addResourcesFromYaml(string(yamlContent))
	if err != nil {
		return nil, err
	}
	return &resources, nil
}

// CreateResourceProviderFromCluster creates a new ResourceProvider using live data from a cluster
func CreateResourceProviderFromCluster(ctx context.Context, c conf.Configuration) (*ResourceProvider, error) {
	dynamicClient, _, clientSet, clusterHost, err := GetKubeClient(ctx, c.KubeContext)
	if err != nil {
		return nil, err
	}
	return CreateResourceProviderFromAPI(ctx, clientSet, clusterHost, dynamicClient, c)
}

func GetKubeClient(ctx context.Context, kubeContext string) (dynamic.Interface, meta.RESTMapper, kubernetes.Interface, string, error) {
	var kubeConf *rest.Config
	var err error
	if len(kubeContext) > 0 {
		kubeConf, err = config.GetConfigWithContext(kubeContext)
	} else {
		kubeConf, err = config.GetConfig()
	}
	if err != nil {
		return nil, nil, nil, "", fmt.Errorf("Error fetching KubeConfig: %v", err)
	}
	clientSet, err := kubernetes.NewForConfig(kubeConf)
	if err != nil {
		return nil, nil, nil, "", fmt.Errorf("Error creating Kubernetes client: %v", err)
	}
	dynamicClient, err := dynamic.NewForConfig(kubeConf)
	if err != nil {
		return nil, nil, nil, "", fmt.Errorf("Error connecting to dynamic interface: %v", err)
	}
	resources, err := restmapper.GetAPIGroupResources(clientSet.Discovery())
	if err != nil {
		return nil, nil, nil, "", fmt.Errorf("Error getting API Group resources: %v", err)
	}
	return dynamicClient, restmapper.NewDiscoveryRESTMapper(resources), clientSet, kubeConf.Host, nil
}

// CreateResourceProviderFromAPI creates a new ResourceProvider from an existing k8s interface
func CreateResourceProviderFromAPI(ctx context.Context, kube kubernetes.Interface, clusterName string, dynamic dynamic.Interface, c conf.Configuration) (*ResourceProvider, error) {
	listOpts := metav1.ListOptions{}
	serverVersion, err := kube.Discovery().ServerVersion()
	if err != nil {
		logrus.Errorf("Error fetching Cluster API version: %v", err)
		return nil, err
	}

	sourceType := "Cluster"
	if c.Namespace != "" {
		logrus.Debug("namespace is specififed in config, setting source type to ClusterNamespace")
		sourceType = "ClusterNamespace"
	}
	provider := newResourceProvider(serverVersion.Major+"."+serverVersion.Minor, sourceType, clusterName)

	logrus.Info("Loading nodes")
	nodes, err := kube.CoreV1().Nodes().List(ctx, listOpts)
	if err != nil {
		logrus.Errorf("Error fetching Nodes: %v", err)
		return nil, err
	}

	logrus.Info("Loading namespaces")
	var namespaces *corev1.NamespaceList
	if c.Namespace != "" {
		ns, err := kube.CoreV1().Namespaces().Get(ctx, c.Namespace, metav1.GetOptions{})
		if err != nil {
			return nil, err
		}
		namespaces = &corev1.NamespaceList{
			Items: []corev1.Namespace{*ns},
		}
	} else {
		nsList, err := kube.CoreV1().Namespaces().List(ctx, listOpts)
		if err != nil {
			logrus.Errorf("Error fetching Namespaces: %v", err)
			return nil, err
		}
		namespaces = nsList
	}

	logrus.Info("Loading pods")
	pods, err := kube.CoreV1().Pods(c.Namespace).List(ctx, listOpts)
	if err != nil {
		logrus.Errorf("Error fetching Pods: %v", err)
		return nil, err
	}

	logrus.Info("Setting up restmapper")
	resources, err := restmapper.GetAPIGroupResources(kube.Discovery())
	if err != nil {
		logrus.Errorf("Error getting API Group resources: %v", err)
		return nil, err
	}
	restMapper := restmapper.NewDiscoveryRESTMapper(resources)
	allChecks := []conf.SchemaCheck{}
	for _, check := range c.CustomChecks {
		allChecks = append(allChecks, check)
	}
	for _, check := range conf.BuiltInChecks {
		allChecks = append(allChecks, check)
	}

	var additionalKinds []conf.TargetKind
	for _, check := range allChecks {
		neededKinds := []conf.TargetKind{check.Target}
		for key := range check.AdditionalSchemas {
			neededKinds = append(neededKinds, conf.TargetKind(key))
		}
		for key := range check.AdditionalSchemaStrings {
			neededKinds = append(neededKinds, conf.TargetKind(key))
		}
		for _, kind := range neededKinds {
			if !funk.Contains(conf.HandledTargets, kind) && !funk.Contains(additionalKinds, kind) {
				additionalKinds = append(additionalKinds, kind)
			}
		}
	}

	var kubernetesResources []GenericResource
	for _, kind := range additionalKinds {
		groupKind := parseGroupKind(maybeTransformKindIntoGroupKind(string(kind)))
		mapping, err := restMapper.RESTMapping(groupKind)
		if err != nil {
			logrus.Warnf("error retrieving mapping of Kind %s because of error: %v", kind, err)
			return nil, err
		}
		if c.Namespace != "" && mapping.Scope.Name() != meta.RESTScopeNameNamespace {
			logrus.Infof("Skipping %s because of auditing specific namespace", mapping.GroupVersionKind)
			continue
		}

		logrus.Info("Loading " + kind)
		objects, err := dynamic.Resource(mapping.Resource).Namespace(c.Namespace).List(ctx, metav1.ListOptions{})
		if err != nil {
			logrus.Warnf("error retrieving parent object API %s and Kind %s because of error: %v", mapping.Resource.Version, mapping.Resource.Resource, err)
			return nil, err
		}
		for _, obj := range objects.Items {
			res, err := NewGenericResourceFromUnstructured(obj, nil)
			if err != nil {
				return nil, err
			}
			kubernetesResources = append(kubernetesResources, res)
		}
	}
	logrus.Info("Loading controllers")
	client := controller.Client{
		Context:    ctx,
		Dynamic:    dynamic,
		RESTMapper: restMapper,
	}
	topControllers, err := client.GetAllTopControllersSummary("")
	if err != nil {
		return nil, fmt.Errorf("error while getting all TopControllers: %v", err)
	}
	for _, workload := range topControllers {
		topController := workload.TopController
		workloadObj, err := NewGenericResourceFromUnstructured(topController, nil)
		if err != nil {
			return nil, fmt.Errorf("could not parse workload %v: %w", workload, err)
		}
		kubernetesResources = append(kubernetesResources, workloadObj)
	}

	provider.Nodes = nodes.Items
	provider.Namespaces = namespaces.Items
	provider.Pods = pods.Items
	provider.Resources.addResources(kubernetesResources)
	logrus.Info("Done loading Kubernetes resources")
	return &provider, nil
}

func (resources *ResourceProvider) addResourcesFromReader(reader io.Reader) error {
	contents, err := io.ReadAll(reader)
	if err != nil {
		logrus.Errorf("Error reading from %v: %v", reader, err)
		return err
	}
	if err := resources.addResourcesFromYaml(string(contents)); err != nil {
		return err
	}
	return nil
}

func (resources *ResourceProvider) addResourcesFromYaml(contents string) error {
	specs := regexp.MustCompile("[\r\n]-+[\r\n]").Split(string(contents), -1)
	for _, spec := range specs {
		if strings.TrimSpace(spec) == "" {
			continue
		}
		err := resources.addResourceFromString(spec)
		if err != nil {
			logrus.Errorf("Error parsing YAML: (%v)", err)
			return err
		}
	}
	return nil
}

func (resources *ResourceProvider) addResourceFromString(contents string) error {
	contentBytes := []byte(contents)
	decoder := k8sYaml.NewYAMLOrJSONDecoder(bytes.NewReader(contentBytes), 1000)
	resource := k8sResource{}
	err := decoder.Decode(&resource)
	decoder = k8sYaml.NewYAMLOrJSONDecoder(bytes.NewReader(contentBytes), 1000)

	if err != nil {
		return err
	}
	if resource.Kind == "Namespace" {
		ns := corev1.Namespace{}
		err = decoder.Decode(&ns)
		resources.Namespaces = append(resources.Namespaces, ns)
	}

	if resource.Kind == "Pod" {
		pod := corev1.Pod{}
		err = decoder.Decode(&pod)
		if err != nil {
			return err
		}
		workload, err := NewGenericResourceFromPod(pod, pod)
		if err != nil {
			return err
		}
		workload.OriginalObjectYAML = contentBytes
		resources.Pods = append(resources.Pods, pod)
		resources.Resources.addResource(workload)
	} else {
		newResource, err := NewGenericResourceFromBytes(contentBytes)
		if err != nil {
			return err
		}
		resources.Resources.addResource(newResource)
	}
	return err
}

// SerializePodSpec converts a typed PodSpec into a map[string]interface{}
func SerializePodSpec(pod *corev1.PodSpec) (map[string]any, error) {
	podJSON, err := json.Marshal(pod)
	if err != nil {
		return nil, err
	}
	podMap := make(map[string]any)
	err = json.Unmarshal(podJSON, &podMap)
	if err != nil {
		return nil, err
	}
	return podMap, nil
}

// SerializePod converts a typed Pod into a map[string]interface{}
func SerializePod(pod *corev1.Pod) (map[string]any, error) {
	podJSON, err := json.Marshal(pod)
	if err != nil {
		return nil, err
	}
	podMap := make(map[string]any)
	err = json.Unmarshal(podJSON, &podMap)
	if err != nil {
		return nil, err
	}
	return podMap, nil
}

// SerializeContainer converts a typed Container into a map[string]interface{}
func SerializeContainer(container *corev1.Container) (map[string]any, error) {
	containerJSON, err := json.Marshal(container)
	if err != nil {
		return nil, err
	}
	containerMap := make(map[string]any)
	err = json.Unmarshal(containerJSON, &containerMap)
	if err != nil {
		return nil, err
	}
	return containerMap, nil
}