feat(scalability): adding generic get_kube_config and getting cluster…

…s to check

hardeneks/__init__.py

@@ -2,7 +2,7 @@
 from pathlib import Path
 from pkg_resources import resource_filename
 import tempfile
-import urllib3
+
 import yaml
 
 from botocore.exceptions import EndpointConnectionError
@@ -16,6 +16,7 @@
     Resources,
 )
 from .harden import harden
+from hardeneks import helpers
 
 
 app = typer.Typer()
@@ -66,14 +67,10 @@ def _get_cluster_name(context, region):
 def _get_region():
     return boto3.session.Session().region_name
 
-
-def _load_kube_config():
-    urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
-    kube_config_orig = f"{Path.home()}/.kube/config"
+def _add_tls_verify():
+    kubeconfig = helpers.get_kube_config()
     tmp_config = tempfile.NamedTemporaryFile().name
 
-    with open(kube_config_orig, "r") as fd:
-        kubeconfig = yaml.safe_load(fd)
     for cluster in kubeconfig["clusters"]:
         cluster["cluster"]["insecure-skip-tls-verify"] = True
     with open(tmp_config, "w") as fd:
@@ -133,8 +130,9 @@ def run_hardeneks(
 
     """
     if insecure_skip_tls_verify:
-        _load_kube_config()
+        _add_tls_verify()
     else:
+        # should pass in config file 
         kubernetes.config.load_kube_config(context=context)
 
     context = _get_current_context(context)

hardeneks/cluster_wide/scalability/control_plane.py

@@ -1,9 +1,10 @@
 import re
-from rich.panel import Panel
+import urllib3
 import kubernetes
-
+from rich.panel import Panel
+from hardeneks import helpers
 from hardeneks import console
-from ...resources import Resources
+from hardeneks import Resources
 
 
 def check_EKS_version(resources: Resources):
@@ -23,17 +24,23 @@ def check_EKS_version(resources: Resources):
 
     return True
 
-
+#
+# check_kubectl_compression
+# checks all clusters in config for disable-compression flag set to true
+# if any cluster does not have setting, it returns False
 def check_kubectl_compression(resources: Resources):
-    _, active_context = kubernetes.config.list_kube_config_contexts()
-    if active_context.get("context", {}).get("disable-compression") != True:
-        console.print(
-            Panel(
-                f"[red]Disable kubectl Compression should equal True",
-                subtitle="[link=https://aws.github.io/aws-eks-best-practices/scalability/docs/control-plane/#disable-kubectl-compression]Click to see the guide[/link]",
+    kubeconfig = helpers.get_kube_config()
+    isSetCorrectly = True
+    for cluster in kubeconfig.get("clusters", []):
+        clusterName = cluster.get("name", "NoName")
+        if cluster.get("cluster", {}).get("disable-compression", False) != True:
+            isSetCorrectly = False
+            console.print(
+                Panel(
+                    f"[red]DisableCompression in Cluster {clusterName}  should equal True",
+                    subtitle="[link=https://aws.github.io/aws-eks-best-practices/scalability/docs/control-plane/#disable-kubectl-compression]Click to see the guide[/link]",
+                )
             )
-        )
-        console.print()
-        return False
+            console.print()
 
-    return True
+    return isSetCorrectly

hardeneks/helpers.py

@@ -0,0 +1,17 @@
+from pathlib import Path
+import urllib3
+import yaml
+
+#
+# get_kube_config
+# returns kube config in json
+# 
+# we need to update this function to take in a config string, so users can pass in kubeconfig as a param
+def get_kube_config():
+    urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
+    # need to fix this, so user can pass in .kube/config as a param (joshkurz)
+    kube_config_orig = f"{Path.home()}/.kube/config"
+
+    with open(kube_config_orig, "r") as fd:
+        kubeconfig = yaml.safe_load(fd)
+    return kubeconfig

hardeneks/resources.py

@@ -1,6 +1,5 @@
 from kubernetes import client
 
-
 class Resources:
     def __init__(self, region, context, cluster, namespaces):
         self.region = region
@@ -68,4 +67,4 @@ def set_resources(self):
             client.AutoscalingV1Api()
             .list_namespaced_horizontal_pod_autoscaler(self.namespace)
             .items
-        )
+        )

tests/test_scalability_control_plane.py

@@ -1,5 +1,6 @@
 from hardeneks.resources import Resources
 from unittest.mock import patch
+from hardeneks import helpers
 
 from hardeneks.cluster_wide.scalability.control_plane import (
     check_EKS_version,
@@ -25,16 +26,20 @@ def test_check_EKS_version(mocked_client):
     mocked_client.return_value = Version("24")
     assert check_EKS_version(namespaced_resources)
 
-@patch("kubernetes.config.list_kube_config_contexts")
-def test_check_kubectl_compression(mocked_client):
+@patch(helpers.__name__ + ".get_kube_config")
+def test_check_kubectl_compression(mocked_helpers):
     namespaced_resources = Resources(
         "some_region", "some_context", "some_cluster", []
     )
-    mocked_client.return_value = None, {'context': {'cluster': 'test', 'user': 'foo', 'disable-compression': True}, 'name': 'foobarcluster'}
+    mocked_helpers.return_value = {'clusters': [{'cluster': {'server': 'testtest', 'disable-compression': True}, 'name': 'foobarcluster'}]}
     assert check_kubectl_compression(namespaced_resources)
-    mocked_client.return_value = None, {'context': {'cluster': 'test', 'user': 'foo'}, 'name': 'foobarcluster'}
+    mocked_helpers.return_value = {'clusters': [{'cluster': {'server': 'testtest', 'disable-compression': True}, 'name': 'foobarcluster6'},{'cluster': {'server': 'testtest', 'disable-compression': True}, 'name': 'foobarcluster2'}]}
+    assert check_kubectl_compression(namespaced_resources)
+    mocked_helpers.return_value = {'clusters': [{'cluster': {'server': 'testtest', 'disable-compression': True}, 'name': 'foobarcluster3'}, {'cluster': {'server': 'testtest', 'disable-compression': False}, 'name': 'foobarcluster4'}]}
+    assert not check_kubectl_compression(namespaced_resources)
+    mocked_helpers.return_value = {'clusters': [{'cluster': {'test': 'user'}, 'name': 'foobarcluster7'}]}
     assert not check_kubectl_compression(namespaced_resources)
-    mocked_client.return_value = None, {'name': 'foobarcluster'}
+    mocked_helpers.return_value = {'clusters': [{}]}
     assert not check_kubectl_compression(namespaced_resources)
-    mocked_client.return_value = None, {'context': {'cluster': 'test', 'user': 'foo', 'disable-compression': False}, 'name': 'foobarcluster'}
-    assert not check_kubectl_compression(namespaced_resources)
+    mocked_helpers.return_value = {}
+    assert check_kubectl_compression(namespaced_resources)