refactor: feature flag utility

Checks for feature flags have been unified into a single function.

connaisseur/constants.py

@@ -6,3 +6,6 @@
 # we can't get closer to 30 than 29 without failing to respond within 30 a lot more often
 AIO_TIMEOUT_SECONDS = 29
 SHA256 = "sha256"
+DETECTION_MODE = "DETECTION_MODE"
+AUTOMATIC_CHILD_APPROVAL = "AUTOMATIC_CHILD_APPROVAL"
+AUTOMATIC_UNCHANGED_APPROVAL = "AUTOMATIC_UNCHANGED_APPROVAL"

connaisseur/exceptions.py

@@ -1,5 +1,7 @@
 import os
 
+import connaisseur.constants as const
+
 
 class BaseConnaisseurException(Exception):
     """
@@ -13,7 +15,10 @@ class BaseConnaisseurException(Exception):
 
     def __init__(self, message: str = default_message, **kwargs):
         self.message = message.format(**kwargs)
-        self.detection_mode = os.environ.get("DETECTION_MODE", "0") == "1"
+        # can't use feature flag, because of circular import
+        self.detection_mode = (
+            os.environ.get(const.DETECTION_MODE, "false").lower() == "true"
+        )
         self.context = {**kwargs, "detection_mode": self.detection_mode}
         super().__init__()
 

connaisseur/flask_application.py

@@ -1,6 +1,5 @@
 import asyncio
 import logging
-import os
 import traceback
 
 import aiohttp
@@ -16,15 +15,14 @@
     BaseConnaisseurException,
     ConfigurationError,
 )
-from connaisseur.util import get_admission_review
+from connaisseur.util import feature_flag_on, get_admission_review
 
 APP = Flask(__name__)
 """
 Flask application that admits the request send to the k8s cluster, validates it and
 sends its response back.
 """
 CONFIG = Config()
-DETECTION_MODE = os.environ.get("DETECTION_MODE", "0") == "1"
 
 metrics = PrometheusMetrics(
     APP,
@@ -129,9 +127,7 @@ async def __async_mutate():
             dispatch_alerts(admission_request, False, msg)
             logging.error(err_log)
             uid = admission_request.uid if admission_request else ""
-            return jsonify(
-                get_admission_review(uid, False, msg=msg, detection_mode=DETECTION_MODE)
-            )
+            return jsonify(get_admission_review(uid, False, msg=msg))
 
 
 async def __admit(admission_request: AdmissionRequest, session: aiohttp.ClientSession):
@@ -167,8 +163,10 @@ async def __validate_image(
     # if automatic_unchanged_approval is enabled, admit resource updates
     # if they do not alter the resource's image reference(s)
     # https://github.com/sse-secure-systems/connaisseur/issues/820
-    unchanged_approval_on = os.environ.get("AUTOMATIC_UNCHANGED_APPROVAL", "0") == "1"
-    if unchanged_approval_on and admission_request.operation.upper() == "UPDATE":
+    if (
+        feature_flag_on(const.AUTOMATIC_UNCHANGED_APPROVAL)
+        and admission_request.operation.upper() == "UPDATE"
+    ):
         old_images = admission_request.old_wl_object.containers.values()
         if image in old_images:
             logging.info(
@@ -183,9 +181,7 @@ async def __validate_image(
     # lookups for already approved images. so child resources are automatically
     # approved without further check ups, if their parents were approved
     # earlier.
-    child_approval_on = os.environ.get("AUTOMATIC_CHILD_APPROVAL_ENABLED", "1") == "1"
-
-    if child_approval_on & (
+    if feature_flag_on(const.AUTOMATIC_CHILD_APPROVAL) and (
         image in admission_request.wl_object.parent_containers.values()
     ):
         logging.info(

connaisseur/util.py

@@ -7,7 +7,8 @@
 import yaml
 from jsonschema import FormatChecker, ValidationError, validate
 
-from connaisseur.exceptions import PathTraversalError
+import connaisseur.constants as const
+from connaisseur.exceptions import BaseConnaisseurException, PathTraversalError
 
 
 def safe_path_func(callback: callable, base_dir: str, path: str, *args, **kwargs):
@@ -32,7 +33,6 @@ def get_admission_review(
     allowed: bool,
     patch: Optional[list] = None,
     msg: Optional[str] = None,
-    detection_mode: bool = False,
 ):
     """
     Get a standardized response object with patching instructions for the
@@ -51,9 +51,6 @@ def get_admission_review(
     msg : str (optional)
         The error message, which will be displayed, should allowed be
         'False'.
-    detection_mode : bool (optional)
-        If set to True, Connaisseur will admit images even if they fail
-        validation, but will log a warning instead.
 
     Return
     ----------
@@ -77,6 +74,7 @@ def get_admission_review(
           }
         }
     """
+    detection_mode = feature_flag_on(const.DETECTION_MODE)
     _, minor, _ = get_kube_version()
     api = "v1beta1" if int(minor) < 17 else "v1"
     review = {
@@ -130,3 +128,14 @@ def get_kube_version():
     regex = r"v(\d)\.(\d{1,2})\.(.*)"
     match = re.match(regex, version)
     return match.groups() if match else ("0", "0", "0")
+
+
+def feature_flag_on(flag: str):
+    env = os.environ.get(flag, "")
+
+    if env.lower() == "true":
+        return True
+    elif env.lower() == "false":
+        return False
+
+    raise BaseConnaisseurException(f"No or wrong value for feature {flag}.")

docs/migrating_to_version_3.md

@@ -49,9 +49,11 @@ Here's the list of changes we made to the Helm `values.yaml`:
     The idea was based around whether the feature would have further configuration or not.
     However, having configuration for a disabled feature doesn't really make sense, so we changed it to be the following:
         - If a feature has configuration, i.e. `<featureName>.<someConfigKey>` is set, it is enabled.
-        - If a feature is set, i.e. `<featureName>: true`  it is enabled and an error is thrown if it needed configuration but doesn't have any.
-        - If a feature is false, i.e. `<featureName>: false`  it is disabled.
+        - If a feature is set, i.e. `<featureName>: true` it is enabled and an error is thrown if it needs configuration but doesn't have any.
+        - If a feature is false, i.e. `<featureName>: false` it is disabled.
         - If a feature is not set at all, i.e. `features` doesn't have a `<featureName>` key, its default configuration is chosen.
+        - If a feature has no configuration, i.e. `<featureName>: true`, the only allowed values are `true` or `false`.
+        Otherwise an error is thrown at time of deployment.
 - Alerting `receivers`
     - We renamed `alerting.templates` to `alerting.receivers` as each receiver has inherently a template and multiple receivers can have the same template.
     As such `templates` wasn't a fitting name.

helm/templates/_helpers.tpl

@@ -217,3 +217,11 @@ Extract Kubernetes Minor Version.
   readOnly: true
 {{- end -}}
 {{- end -}}
+
+{{- define "checkFalseTrue" -}}
+{{- if (eq .input "true") | or (eq .input "false") -}}
+{{ .input }}
+{{- else -}}
+{{- fail "wrong value" -}}
+{{- end -}}
+{{- end -}}

helm/templates/env.yaml

@@ -10,20 +10,14 @@ data:
   KUBE_API_TOKEN_PATH: /var/run/secrets/kubernetes.io/serviceaccount/token
   KUBE_API_CA_PATH: /var/run/secrets/kubernetes.io/serviceaccount/ca.crt
   LOG_LEVEL: {{.Values.application.logLevel | default "INFO"}}
-  {{- if .Values.application.features.detectionMode }}
-  DETECTION_MODE: "1"
-  {{- end }}
-  {{- if .Values.application.features.automaticChildApproval }}
-  AUTOMATIC_CHILD_APPROVAL_ENABLED: "1"
-  {{- else }}
-  AUTOMATIC_CHILD_APPROVAL_ENABLED: "0"
+  {{- with .Values.application.features }}
+  DETECTION_MODE: {{ print .detectionMode | default "false" | dict "input" | include "checkFalseTrue" | quote }}
+  AUTOMATIC_CHILD_APPROVAL: {{ print .automaticChildApproval | default "true" | dict "input" | include "checkFalseTrue" | quote }}
+  AUTOMATIC_UNCHANGED_APPROVAL: {{ print .automaticUnchangedApproval | default "true" | dict "input" | include "checkFalseTrue" | quote }}
   {{- end }}
   {{- if .Values.alerting }}
   CLUSTER_NAME: {{ default "not specified" .Values.alerting.clusterIdentifier }}
   {{- end }}
-  {{- if .Values.application.features.automaticUnchangedApproval  }}
-  AUTOMATIC_UNCHANGED_APPROVAL: "1"
-  {{- end }}
   KUBE_VERSION: {{ .Capabilities.KubeVersion }}
 ---
 apiVersion: v1

scripts/upgrade_to_version_3.py

@@ -41,15 +41,15 @@
     validators = config.pop("validators")
     policy = config.pop("policy")
 
-    detection_mode = config.pop("detectionMode", False)
+    detection_mode = bool(config.pop("detectionMode", False))
     namespaced_validation = config.pop("namespacedValidation", {"enabled": False})
     # check explicitly disabled config won't be taken as enabled
     if not namespaced_validation.pop("enabled"):
         namespaced_validation = False
-    child_approval = config.pop("automaticChildApproval", {"enabled": True}).get(
-        "enabled"
+    child_approval = bool(
+        config.pop("automaticChildApproval", {"enabled": True}).get("enabled")
     )
-    unchanged_approval = config.pop("automaticUnchangedApproval", False)
+    unchanged_approval = bool(config.pop("automaticUnchangedApproval", False))
     features = {
         "detectionMode": detection_mode,
         "namespacedValidation": namespaced_validation,

tests/conftest.py

@@ -366,7 +366,7 @@ def m_safe_path_func(monkeypatch):
 
 @pytest.fixture
 def m_alerting(monkeypatch, m_safe_path_func):
-    monkeypatch.setenv("DETECTION_MODE", "0")
+    monkeypatch.setenv("DETECTION_MODE", "false")
     monkeypatch.setenv("POD_NAME", "connaisseur-pod-123")
     monkeypatch.setenv("CLUSTER_NAME", "minikube")
     connaisseur.alert.AlertingConfiguration._AlertingConfiguration__PATH = (
@@ -380,7 +380,7 @@ def m_alerting(monkeypatch, m_safe_path_func):
 
 @pytest.fixture
 def m_alerting_without_send(monkeypatch, m_safe_path_func, mocker):
-    monkeypatch.setenv("DETECTION_MODE", "0")
+    monkeypatch.setenv("DETECTION_MODE", "false")
     monkeypatch.setenv("POD_NAME", "connaisseur-pod-123")
     monkeypatch.setenv("CLUSTER_NAME", "minikube")
     monkeypatch.setattr(

tests/test_exceptions.py

@@ -8,16 +8,16 @@
 @pytest.mark.parametrize(
     "message, kwargs, detection_mode, ex_msg",
     [
-        ("Error.", {"reason": "fate"}, 0, "Error."),
+        ("Error.", {"reason": "fate"}, False, "Error."),
         (
             "How the {tables} have {turned}.",
             {"tables": "turns", "turned": "tabled"},
-            0,
+            False,
             "How the turns have tabled.",
         ),
-        ("", {}, 0, "An error occurred."),
-        ("", {"bad": "guy"}, 0, "An error occurred."),
-        ("", {}, 1, "An error occurred."),
+        ("", {}, False, "An error occurred."),
+        ("", {"bad": "guy"}, False, "An error occurred."),
+        ("", {}, True, "An error occurred."),
     ],
 )
 def test_exceptions(monkeypatch, message, kwargs, detection_mode, ex_msg):
@@ -28,7 +28,7 @@ def test_exceptions(monkeypatch, message, kwargs, detection_mode, ex_msg):
         ex = exc.BaseConnaisseurException(**kwargs)
     assert ex.message == ex_msg
     assert ex.context == dict(**kwargs, detection_mode=ex.detection_mode)
-    assert ex.detection_mode is bool(detection_mode)
+    assert ex.detection_mode == detection_mode
 
 
 def test_exception_str():
@@ -44,8 +44,8 @@ def test_exception_str():
 @pytest.mark.parametrize(
     "msg, dm, out",
     [
-        ("hello there.", "0", "hello there."),
-        ("hello there.", "1", "hello there. (not denied due to DETECTION_MODE)"),
+        ("hello there.", "false", "hello there."),
+        ("hello there.", "true", "hello there. (not denied due to DETECTION_MODE)"),
     ],
 )
 def test_exception_user_msg(monkeypatch, msg, dm, out):

tests/test_flask_application.py

@@ -67,9 +67,16 @@ def mock_init(self):
     pytest.fa = fa
 
 
+@pytest.fixture(autouse=True)
+def set_envs(monkeypatch):
+    monkeypatch.setenv("AUTOMATIC_CHILD_APPROVAL", "true")
+    monkeypatch.setenv("AUTOMATIC_UNCHANGED_APPROVAL", "true")
+    monkeypatch.setenv("DETECTION_MODE", "false")
+
+
 @pytest.mark.parametrize(
     "index, allowed, status_code, detection_mode",
-    [(0, True, 202, 0), (5, False, 403, 0)],
+    [(0, True, 202, False), (5, False, 403, False)],
 )
 def test_mutate(
     monkeypatch,
@@ -101,7 +108,7 @@ def test_mutate_calls_send_alert_for_invalid_admission_request(
 ):
     with aioresponses() as aio:
         aio.get(re.compile(r".*"), callback=fix.async_callback, repeat=True)
-        monkeypatch.setenv("DETECTION_MODE", "0")
+        monkeypatch.setenv("DETECTION_MODE", "false")
         client = pytest.fa.APP.test_client()
         response = client.post("/mutate", json=adm_req_samples[7])
         admission_response = response.get_json()["response"]
@@ -206,7 +213,7 @@ async def test_admit(
     session = aiohttp.ClientSession()
     with exception:
         if index == 8:
-            monkeypatch.setenv("AUTOMATIC_UNCHANGED_APPROVAL", "1")
+            monkeypatch.setenv("AUTOMATIC_UNCHANGED_APPROVAL", "true")
         with aioresponses() as aio:
             aio.get(re.compile(r".*"), callback=fix.async_callback, repeat=True)
             response = await pytest.fa.__admit(

tests/test_util.py

@@ -101,10 +101,8 @@ def test_safe_path_func(path, exception):
 )
 def test_get_admission_review(monkeypatch, uid, allowed, patch, msg, dm, review):
     monkeypatch.setenv("KUBE_VERSION", "v1.20.0")
-    assert (
-        ut.get_admission_review(uid, allowed, patch=patch, msg=msg, detection_mode=dm)
-        == review
-    )
+    monkeypatch.setenv("DETECTION_MODE", str(dm))
+    assert ut.get_admission_review(uid, allowed, patch=patch, msg=msg) == review
 
 
 @pytest.mark.parametrize(
@@ -145,3 +143,34 @@ def test_validate_schema(
 def test_get_kube_version(monkeypatch, major, minor, patch, set_version):
     monkeypatch.setenv("KUBE_VERSION", set_version)
     assert ut.get_kube_version() == (major, minor, patch)
+
+
+@pytest.mark.parametrize(
+    "flag, value, expected_value, exception",
+    [
+        ("FLAG", True, True, fix.no_exc()),
+        ("FLAG", "True", True, fix.no_exc()),
+        ("FLAG", "False", False, fix.no_exc()),
+        (
+            "FLAG",
+            None,
+            False,
+            pytest.raises(
+                exc.BaseConnaisseurException, match=r"No or wrong value for feature .*"
+            ),
+        ),
+        (
+            "FLAG",
+            "excuse_me_?",
+            False,
+            pytest.raises(
+                exc.BaseConnaisseurException, match=r"No or wrong value for feature .*"
+            ),
+        ),
+    ],
+)
+def test_feature_flag_on(monkeypatch, flag, value, expected_value, exception):
+    with exception:
+        if value is not None:
+            monkeypatch.setenv(flag, value)
+        assert ut.feature_flag_on(flag) == expected_value