diff options
Diffstat (limited to 'test/security/check_versions/tests')
6 files changed, 318 insertions, 0 deletions
diff --git a/test/security/check_versions/tests/conftest.py b/test/security/check_versions/tests/conftest.py new file mode 100644 index 000000000..7c3e2e171 --- /dev/null +++ b/test/security/check_versions/tests/conftest.py @@ -0,0 +1,12 @@ +#!/usr/bin/env python3 + +import pytest + + +def pod_name_trimmer_fun(pod_name): + return "-".join(pod_name.split("-")[:-2]) + + +@pytest.fixture +def pod_name_trimmer(): + return pod_name_trimmer_fun diff --git a/test/security/check_versions/tests/test_gather_containers_informations.py b/test/security/check_versions/tests/test_gather_containers_informations.py new file mode 100644 index 000000000..63401721e --- /dev/null +++ b/test/security/check_versions/tests/test_gather_containers_informations.py @@ -0,0 +1,38 @@ +#!/usr/bin/env python3 + +import k8s_bin_versions_inspector as kbvi +import kubernetes + + +def test_gather_containers_informations(pod_name_trimmer): + kubernetes.config.load_kube_config() + api = kubernetes.client.CoreV1Api() + containers = kbvi.gather_containers_informations(api, "", False) + data = [ + ( + c.namespace, + pod_name_trimmer(c.pod), + c.container, + c.versions.python, + c.versions.java, + ) + for c in containers + ] + sorted_data = sorted(data) + assert sorted_data == [ + ("default", "kbvi-test-java-keycloak", "keycloak", [], ["11.0.8"]), + ("default", "kbvi-test-java-keycloak-old", "keycloak-old", [], ["11.0.5"]), + ( + "default", + "kbvi-test-java-keycloak-very-old", + "keycloak-very-old", + ["2.7.5"], + [], + ), # TODO + ("default", "kbvi-test-python-jupyter", "jupyter", ["3.8.4"], []), + ("default", "kbvi-test-python-jupyter-old", "jupyter-old", ["3.6.6"], []), + ("default", "kbvi-test-python-stderr-filebeat", "filebeat", ["2.7.5"], []), + ("default", "kbvi-test-terminated", "python", [], []), # TODO + ("ingress-nginx", "kbvi-test-ingress-nginx", "echo-server", [], []), + ("kube-system", "kbvi-test-kube-system", "echo-server", [], []), + ] diff --git a/test/security/check_versions/tests/test_list_all_containers.py b/test/security/check_versions/tests/test_list_all_containers.py new file mode 100644 index 000000000..4178077c3 --- /dev/null +++ b/test/security/check_versions/tests/test_list_all_containers.py @@ -0,0 +1,52 @@ +#!/usr/bin/env python3 + +import k8s_bin_versions_inspector as kbvi +import kubernetes + + +def exec_list_all_containers(pod_name_trimmer, field_selector): + kubernetes.config.load_kube_config() + api = kubernetes.client.CoreV1Api() + containers = kbvi.list_all_containers(api, field_selector) + extracted = ((c.namespace, c.pod, c.container) for c in containers) + trimmed = ((n, pod_name_trimmer(p), c) for n, p, c in extracted) + result = sorted(trimmed) + return result + + +def test_list_all_containers(pod_name_trimmer): + result = exec_list_all_containers(pod_name_trimmer, "") + assert result == [ + ("default", "kbvi-test-java-keycloak", "keycloak"), + ("default", "kbvi-test-java-keycloak-old", "keycloak-old"), + ("default", "kbvi-test-java-keycloak-very-old", "keycloak-very-old"), + ("default", "kbvi-test-python-jupyter", "jupyter"), + ("default", "kbvi-test-python-jupyter-old", "jupyter-old"), + ("default", "kbvi-test-python-stderr-filebeat", "filebeat"), + ("default", "kbvi-test-terminated", "python"), + ("ingress-nginx", "kbvi-test-ingress-nginx", "echo-server"), + ("kube-system", "kbvi-test-kube-system", "echo-server"), + ] + + +def test_list_all_containers_not_default(pod_name_trimmer): + field_selector = "metadata.namespace!=default" + result = exec_list_all_containers(pod_name_trimmer, field_selector) + assert result == [ + ("ingress-nginx", "kbvi-test-ingress-nginx", "echo-server"), + ("kube-system", "kbvi-test-kube-system", "echo-server"), + ] + + +def test_list_all_containers_conjunction(pod_name_trimmer): + field_selector = "metadata.namespace!=kube-system,metadata.namespace!=ingress-nginx" + result = exec_list_all_containers(pod_name_trimmer, field_selector) + assert result == [ + ("default", "kbvi-test-java-keycloak", "keycloak"), + ("default", "kbvi-test-java-keycloak-old", "keycloak-old"), + ("default", "kbvi-test-java-keycloak-very-old", "keycloak-very-old"), + ("default", "kbvi-test-python-jupyter", "jupyter"), + ("default", "kbvi-test-python-jupyter-old", "jupyter-old"), + ("default", "kbvi-test-python-stderr-filebeat", "filebeat"), + ("default", "kbvi-test-terminated", "python"), + ] diff --git a/test/security/check_versions/tests/test_main.py b/test/security/check_versions/tests/test_main.py new file mode 100644 index 000000000..0dff0b230 --- /dev/null +++ b/test/security/check_versions/tests/test_main.py @@ -0,0 +1,84 @@ +#!/usr/bin/env python3 + +import k8s_bin_versions_inspector as kbvi +import json +import tempfile +import yaml + + +def exec_main(pod_name_trimmer, acceptable_data): + + with tempfile.NamedTemporaryFile() as output_temp, tempfile.NamedTemporaryFile() as acceptable_temp: + + with open(acceptable_temp.name, "w") as stream: + yaml.safe_dump(acceptable_data, stream) + + result = kbvi.main( + [ + "--quiet", + "--output-file", + output_temp.name, + "--output-format", + "json", + "--acceptable", + acceptable_temp.name, + ] + ) + + with open(output_temp.name, "r") as stream: + output_data = json.load(stream) + output_extracted = ( + ( + item["namespace"], + pod_name_trimmer(item["pod"]), + item["container"], + item["versions"]["python"], + item["versions"]["java"], + ) + for item in output_data + ) + output_sorted = sorted(output_extracted) + + assert output_sorted == [ + ("default", "kbvi-test-java-keycloak", "keycloak", [], ["11.0.8"]), + ("default", "kbvi-test-java-keycloak-old", "keycloak-old", [], ["11.0.5"]), + ( + "default", + "kbvi-test-java-keycloak-very-old", + "keycloak-very-old", + ["2.7.5"], + [], + ), + ("default", "kbvi-test-python-jupyter", "jupyter", ["3.8.4"], []), + ("default", "kbvi-test-python-jupyter-old", "jupyter-old", ["3.6.6"], []), + ("default", "kbvi-test-python-stderr-filebeat", "filebeat", ["2.7.5"], []), + ("default", "kbvi-test-terminated", "python", [], []), + ("ingress-nginx", "kbvi-test-ingress-nginx", "echo-server", [], []), + ("kube-system", "kbvi-test-kube-system", "echo-server", [], []), + ] + + return result + + +def test_main(pod_name_trimmer): + + acceptable_data = { + "python": ["2.7.5", "3.6.6", "3.8.4"], + "java": ["11.0.5", "11.0.8"], + } + + result = exec_main(pod_name_trimmer, acceptable_data) + + assert result == 0 + + +def test_main_neg(pod_name_trimmer): + + acceptable_data = { + "python": ["3.6.6", "3.8.4"], + "java": ["11.0.5", "11.0.8"], + } + + result = exec_main(pod_name_trimmer, acceptable_data) + + assert result == 1 diff --git a/test/security/check_versions/tests/test_sync_post_namespaced_pod_exec.py b/test/security/check_versions/tests/test_sync_post_namespaced_pod_exec.py new file mode 100644 index 000000000..50620d3a7 --- /dev/null +++ b/test/security/check_versions/tests/test_sync_post_namespaced_pod_exec.py @@ -0,0 +1,74 @@ +#!/usr/bin/env python3 + +import k8s_bin_versions_inspector as kbvi +import kubernetes + + +def exec_sync_post_namespaced_pod_exec(pod, command): + kubernetes.config.load_kube_config() + api = kubernetes.client.CoreV1Api() + containers = kbvi.list_all_containers(api, "") + container = next(c for c in containers if c.pod.startswith(pod)) + result = kbvi.sync_post_namespaced_pod_exec(api, container, command) + return result + + +def test_sync_post_namespaced_pod_exec(): + pod = "kbvi-test-python-jupyter" + result = exec_sync_post_namespaced_pod_exec(pod, "id") + assert result == { + "stdout": "uid=1000(jovyan) gid=100(users) groups=100(users)\n", + "stderr": "", + "error": {"status": "Success", "metadata": {}}, + "code": 0, + } + + +def test_sync_post_namespaced_pod_exec_not_running(): + pod = "kbvi-test-terminated" + result = exec_sync_post_namespaced_pod_exec(pod, "id") + assert result == {"stdout": "", "stderr": "", "error": {}, "code": -1} + + +def test_sync_post_namespaced_pod_exec_not_found(): + pod = "kbvi-test-python-jupyter" + command = "/command/not/found" + result = exec_sync_post_namespaced_pod_exec(pod, command) + assert result["stdout"] == "" + assert result["stderr"] == "" + assert result["error"]["status"] == "Failure" + assert result["error"]["reason"] == "InternalError" + assert result["code"] == -2 + + +def test_sync_post_namespaced_pod_exec_exit_code(): + pod = "kbvi-test-python-jupyter" + command = ["python3", "--invalid-attribute"] + result = exec_sync_post_namespaced_pod_exec(pod, command) + assert result == { + "stdout": "", + "stderr": "unknown option --invalid-attribute\n" + "usage: python3 [option] ... [-c cmd | -m mod | file | -] [arg] ...\n" + "Try `python -h' for more information.\n", + "error": { + "status": "Failure", + "reason": "NonZeroExitCode", + "message": "command terminated with non-zero exit code: error " + "executing command [python3 --invalid-attribute], exit code 2", + "details": {"causes": [{"message": "2", "reason": "ExitCode"}]}, + "metadata": {}, + }, + "code": 2, + } + + +def test_sync_post_namespaced_pod_exec_stderr(): + pod = "kbvi-test-python-stderr-filebeat" + command = ["python", "--version"] + result = exec_sync_post_namespaced_pod_exec(pod, command) + assert result == { + "stdout": "", + "stderr": "Python 2.7.5\n", + "error": {"status": "Success", "metadata": {}}, + "code": 0, + } diff --git a/test/security/check_versions/tests/test_verify_versions_acceptability.py b/test/security/check_versions/tests/test_verify_versions_acceptability.py new file mode 100644 index 000000000..5e2f0d2c8 --- /dev/null +++ b/test/security/check_versions/tests/test_verify_versions_acceptability.py @@ -0,0 +1,58 @@ +#!/usr/bin/env python3 + +import k8s_bin_versions_inspector as kbvi +import yaml +import tempfile +import pathlib + + +def exec_verify_versions_acceptability(containers): + + config = { + "python": ["1.1.1", "2.2.2"], + "java": ["3.3.3"], + } + + with tempfile.NamedTemporaryFile() as temp: + with open(temp.name, "w") as stream: + yaml.safe_dump(config, stream) + acceptable = pathlib.Path(temp.name) + result = kbvi.verify_versions_acceptability(containers, acceptable, True) + + return result + + +def test_verify_versions_acceptability(): + + containers = [ + kbvi.ContainerInfo("a", "b", "c", None, kbvi.ContainerVersions([], [])), + kbvi.ContainerInfo( + "a", "b", "c", None, kbvi.ContainerVersions(["1.1.1"], ["3.3.3"]) + ), + ] + + result = exec_verify_versions_acceptability(containers) + + assert result == 0 + + +def test_verify_versions_acceptability_neg_1(): + + containers = [ + kbvi.ContainerInfo("a", "b", "c", None, kbvi.ContainerVersions(["3.3.3"], [])) + ] + + result = exec_verify_versions_acceptability(containers) + + assert result == 1 + + +def test_verify_versions_acceptability_neg_2(): + + containers = [ + kbvi.ContainerInfo("a", "b", "c", None, kbvi.ContainerVersions([], ["1.1.1"])) + ] + + result = exec_verify_versions_acceptability(containers) + + assert result == 1 |