[host-sensor] first integration in kubescape

hostsensorutils/hostsensordeploy.go

@@ -1,11 +1,11 @@
 package hostsensorutils
 
 import (
-	"encoding/json"
 	"fmt"
 	"io"
 	"strings"
 	"sync"
+	"time"
 
 	"github.com/armosec/k8s-interface/k8sinterface"
 	appsv1 "k8s.io/api/apps/v1"
@@ -28,12 +28,6 @@ type HostSensorHandler struct {
 	podListLock             sync.RWMutex
 }
 
-type HostSensorDataEnvelope struct {
-	Kind     string            `json:"kind"`
-	NodeName string            `json:"nodeName"`
-	Data     []json.RawMessage `json:"data"`
-}
-
 func NewHostSensorHandler(k8sObj *k8sinterface.KubernetesApi) (*HostSensorHandler, error) {
 	// deploy the YAML
 	// store namespace + port
@@ -47,37 +41,49 @@ func NewHostSensorHandler(k8sObj *k8sinterface.KubernetesApi) (*HostSensorHandle
 	if err := hsh.applyYAML(); err != nil {
 		return nil, fmt.Errorf("in NewHostSensorHandler, failed to apply YAML: %v", err)
 	}
+	hsh.populatePodNamesToNodeNames()
+	if err := hsh.checkPodForEachNode(); err != nil {
+		fmt.Printf("failed to validate host-sensor pods status: %v", err)
+	}
 	return hsh, nil
 }
 
 func (hsh *HostSensorHandler) applyYAML() error {
 	dec := yaml.NewDocumentDecoder(io.NopCloser(strings.NewReader(hostSensorYAML)))
 	// apply namespace
-	singleYAMLBytes := make([]byte, 0, 2048)
-	if _, err := dec.Read(singleYAMLBytes); err != nil {
+	singleYAMLBytes := make([]byte, 4096)
+	if readLen, err := dec.Read(singleYAMLBytes); err != nil {
 		return fmt.Errorf("failed to read YAML of namespace: %v", err)
+	} else {
+		singleYAMLBytes = singleYAMLBytes[:readLen]
 	}
 	namespaceAC := &coreapplyv1.NamespaceApplyConfiguration{}
 	if err := yaml.Unmarshal(singleYAMLBytes, namespaceAC); err != nil {
 		return fmt.Errorf("failed to Unmarshal YAML of namespace: %v", err)
 	}
 
-	if ns, err := hsh.k8sObj.KubernetesClient.CoreV1().Namespaces().Apply(hsh.k8sObj.Context, namespaceAC, metav1.ApplyOptions{}); err != nil {
+	if ns, err := hsh.k8sObj.KubernetesClient.CoreV1().Namespaces().Apply(hsh.k8sObj.Context, namespaceAC, metav1.ApplyOptions{
+		FieldManager: "kubescape",
+	}); err != nil {
 		return fmt.Errorf("failed to apply YAML of namespace: %v", err)
 	} else {
 		hsh.HostSensorNamespace = ns.Name
 	}
 	// apply deamonset
 	deamonAC := &appsapplyv1.DaemonSetApplyConfiguration{}
-	singleYAMLBytes = make([]byte, 0, 4096)
-	if _, err := dec.Read(singleYAMLBytes); err != nil {
+	singleYAMLBytes = make([]byte, 4096)
+	if readLen, err := dec.Read(singleYAMLBytes); err != nil {
 		return fmt.Errorf("failed to read YAML of deamonset: %v", err)
+	} else {
+		singleYAMLBytes = singleYAMLBytes[:readLen]
 	}
 	if err := yaml.Unmarshal(singleYAMLBytes, deamonAC); err != nil {
 		return fmt.Errorf("failed to Unmarshal YAML of deamonset: %v", err)
 	}
 	deamonAC.Namespace = &hsh.HostSensorNamespace
-	if ds, err := hsh.k8sObj.KubernetesClient.AppsV1().DaemonSets(hsh.HostSensorNamespace).Apply(hsh.k8sObj.Context, deamonAC, metav1.ApplyOptions{}); err != nil {
+	if ds, err := hsh.k8sObj.KubernetesClient.AppsV1().DaemonSets(hsh.HostSensorNamespace).Apply(hsh.k8sObj.Context, deamonAC, metav1.ApplyOptions{
+		FieldManager: "kubescape",
+	}); err != nil {
 		return fmt.Errorf("failed to apply YAML of deamonset: %v", err)
 	} else {
 		hsh.HostSensorDaemonSetName = ds.Name
@@ -87,12 +93,34 @@ func (hsh *HostSensorHandler) applyYAML() error {
 	return nil
 }
 
-func (hsh *HostSensorHandler) populatePodNamesToNodeNames() error {
+func (hsh *HostSensorHandler) checkPodForEachNode() error {
+	deadline := time.Now().Add(time.Second * 100)
+	for {
+		nodesList, err := hsh.k8sObj.KubernetesClient.CoreV1().Nodes().List(hsh.k8sObj.Context, metav1.ListOptions{})
+		if err != nil {
+			return fmt.Errorf("in checkPodsForEveryNode, failed to get nodes list: %v", nodesList)
+		}
+		hsh.podListLock.RLock()
+		podsNum := len(hsh.HostSensorPodNames)
+		hsh.podListLock.RUnlock()
+		if len(nodesList.Items) == podsNum {
+			break
+		}
+		if time.Now().After(deadline) {
+			return fmt.Errorf("host-sensor pods number (%d) differ than nodes number (%d) after deadline exceded", podsNum, len(nodesList.Items))
+		}
+		time.Sleep(10 * time.Second)
+	}
+	return nil
+}
+
+// initiating routine to keep pod list updated
+func (hsh *HostSensorHandler) populatePodNamesToNodeNames() {
 
 	go func() {
 		watchRes, err := hsh.k8sObj.KubernetesClient.CoreV1().Pods(hsh.DaemonSet.Namespace).Watch(hsh.k8sObj.Context, metav1.ListOptions{
 			Watch:         true,
-			LabelSelector: fmt.Sprintf("app=%s", hsh.DaemonSet.Labels["app"]),
+			LabelSelector: fmt.Sprintf("name=%s", hsh.DaemonSet.Spec.Template.Labels["name"]),
 		})
 		if err != nil {
 			fmt.Printf("Failed to watch over daemonset pods")
@@ -103,36 +131,38 @@ func (hsh *HostSensorHandler) populatePodNamesToNodeNames() error {
 				fmt.Printf("Failed to watch over daemonset pods: not a Pod")
 				continue
 			}
-			hsh.podListLock.Lock()
-			switch eve.Type {
-			case watch.Added:
-				if pod.Status.Phase == corev1.PodRunning {
-					hsh.HostSensorPodNames[pod.ObjectMeta.Name] = pod.Spec.NodeName
-				} else {
-					delete(hsh.HostSensorPodNames, pod.ObjectMeta.Name)
-				}
-			default:
-				delete(hsh.HostSensorPodNames, pod.ObjectMeta.Name)
-			}
-			hsh.podListLock.Unlock()
+			go hsh.updatePodInListAtomic(eve.Type, pod)
 		}
 	}()
+}
 
-	return nil
+func (hsh *HostSensorHandler) updatePodInListAtomic(eventType watch.EventType, podObj *corev1.Pod) {
+	hsh.podListLock.Lock()
+	defer hsh.podListLock.Unlock()
+
+	switch eventType {
+	case watch.Added:
+		if podObj.Status.Phase == corev1.PodRunning {
+			hsh.HostSensorPodNames[podObj.ObjectMeta.Name] = podObj.Spec.NodeName
+		} else {
+			delete(hsh.HostSensorPodNames, podObj.ObjectMeta.Name)
+		}
+	default:
+		delete(hsh.HostSensorPodNames, podObj.ObjectMeta.Name)
+	}
 }
 
 func (hsh *HostSensorHandler) TearDown() error {
 	// remove the namespace
-	if err := hsh.k8sObj.KubernetesClient.CoreV1().Namespaces().Delete(hsh.k8sObj.Context, hsh.HostSensorNamespace, metav1.DeleteOptions{}); err != nil {
+	gracePeriod := int64(15)
+	if err := hsh.k8sObj.KubernetesClient.AppsV1().DaemonSets(hsh.HostSensorNamespace).Delete(hsh.k8sObj.Context, hsh.HostSensorDaemonSetName, metav1.DeleteOptions{GracePeriodSeconds: &gracePeriod}); err != nil {
+		return fmt.Errorf("failed to delete host-sensor daemonset: %v", err)
+	}
+	if err := hsh.k8sObj.KubernetesClient.CoreV1().Namespaces().Delete(hsh.k8sObj.Context, hsh.HostSensorNamespace,
+		metav1.DeleteOptions{GracePeriodSeconds: &gracePeriod}); err != nil {
 		return fmt.Errorf("failed to delete host-sensor namespace: %v", err)
 	}
 	// TODO: wait for termination
 
 	return nil
 }
-
-// return list of
-func (hsh *HostSensorHandler) GetKubeletConfigurations() ([][]byte, error) {
-	// loop over pods and port-forward it to each of them
-	return make([][]byte, 0), nil
-}

hostsensorutils/hostsensorgetfrompod.go

@@ -0,0 +1,107 @@
+package hostsensorutils
+
+import (
+	"encoding/json"
+	"fmt"
+	"sync"
+
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	"sigs.k8s.io/yaml"
+)
+
+type HostSensorDataEnvelope struct {
+	schema.GroupVersionKind
+	NodeName string          `json:"nodeName"`
+	Data     json.RawMessage `json:"data"`
+}
+
+func (hsh *HostSensorHandler) getPodList() (res map[string]string, err error) {
+	hsh.podListLock.RLock()
+	jsonBytes, err := json.Marshal(hsh.HostSensorPodNames)
+	hsh.podListLock.RUnlock()
+	if err != nil {
+		return res, fmt.Errorf("failed to marshal pod list: %v", err)
+	}
+	err = json.Unmarshal(jsonBytes, &res)
+	if err != nil {
+		return res, fmt.Errorf("failed to unmarshal pod list: %v", err)
+	}
+	return res, nil
+}
+
+func (hsh *HostSensorHandler) HTTPGetToPod(podName, path string) ([]byte, error) {
+	//  send the request to the port
+
+	restProxy := hsh.k8sObj.KubernetesClient.CoreV1().Pods(hsh.DaemonSet.Namespace).ProxyGet("http", podName, fmt.Sprintf("%d", hsh.HostSensorPort), path, map[string]string{})
+	return restProxy.DoRaw(hsh.k8sObj.Context)
+
+}
+
+func (hsh *HostSensorHandler) ForwardToPod(podName, path string) ([]byte, error) {
+	// NOT IN USE:
+	// ---
+	// spawn port forwarding
+	// req := hsh.k8sObj.KubernetesClient.CoreV1().RESTClient().Post()
+	// req = req.Name(podName)
+	// req = req.Namespace(hsh.DaemonSet.Namespace)
+	// req = req.Resource("pods")
+	// req = req.SubResource("portforward")
+	// ----
+	// https://github.com/gianarb/kube-port-forward
+	// fullPath := fmt.Sprintf("/api/v1/namespaces/%s/pods/%s/portforward",
+	// 	hsh.DaemonSet.Namespace, podName)
+	// transport, upgrader, err := spdy.RoundTripperFor(hsh.k8sObj.KubernetesClient.config)
+	// if err != nil {
+	// 	return nil, err
+	// }
+	// hostIP := strings.TrimLeft(req.RestConfig.Host, "htps:/")
+	// dialer := spdy.NewDialer(upgrader, &http.Client{Transport: transport}, http.MethodPost, &url.URL{Scheme: "http", Path: path, Host: hostIP})
+	return nil, nil
+}
+
+// sendAllPodsHTTPGETRequest fills the raw byte response in the envelope and the node name, but not the GroupVersionKind
+// so the caller is responsible to convert the raw data to some structured data and add the GroupVersionKind details
+func (hsh *HostSensorHandler) sendAllPodsHTTPGETRequest(path string) ([]HostSensorDataEnvelope, error) {
+	podList, err := hsh.getPodList()
+	if err != nil {
+		return nil, fmt.Errorf("failed to sendAllPodsHTTPGETRequest: %v", err)
+	}
+	res := make([]HostSensorDataEnvelope, 0, len(podList))
+	resLock := sync.Mutex{}
+	wg := sync.WaitGroup{}
+	wg.Add(len(podList))
+	for podName := range podList {
+		go func(podName, path string) {
+			defer wg.Done()
+			resBytes, err := hsh.HTTPGetToPod(podName, path)
+			if err != nil {
+				fmt.Printf("In sendAllPodsHTTPGETRequest failed to get data '%s' from pod '%s': %v", path, podName, err)
+			} else {
+				resLock.Lock()
+				defer resLock.Unlock()
+				res = append(res, HostSensorDataEnvelope{NodeName: podList[podName], Data: resBytes})
+			}
+
+		}(podName, path)
+	}
+	wg.Wait()
+	return res, nil
+}
+
+// return list of
+func (hsh *HostSensorHandler) GetKubeletConfigurations() ([]HostSensorDataEnvelope, error) {
+	// loop over pods and port-forward it to each of them
+	res, err := hsh.sendAllPodsHTTPGETRequest("/kubeletConfigurations")
+	for resIdx := range res {
+		jsonBytes, err := yaml.YAMLToJSON(res[resIdx].Data)
+		if err != nil {
+			fmt.Printf("In GetKubeletConfigurations failed to YAMLToJSON: %v;\n%v", err, res[resIdx])
+		}
+		res[resIdx].Data = jsonBytes
+		kindDet := schema.GroupVersionKind{}
+		if err = json.Unmarshal(jsonBytes, &kindDet); err != nil {
+			fmt.Printf("In GetKubeletConfigurations failed to Unmarshal GroupVersionKind: %v;\n%v", err, jsonBytes)
+		}
+	}
+	return res, err
+}

hostsensorutils/hostsensoryamls.go

@@ -1,7 +1,6 @@
 package hostsensorutils
 
-const hostSensorYAML = `
-apiVersion: v1
+const hostSensorYAML = `apiVersion: v1
 kind: Namespace
 metadata:
   labels:
@@ -63,8 +62,4 @@ spec:
         name: host-filesystem
       hostNetwork: true
       hostPID: true
-      hostIPC: true
-
-    
-
-      `
+      hostIPC: true`