elasticapm/conf: block disabling of verify cert server in fips mode

Author: xrmx

elasticapm/conf/__init__.py

@@ -37,6 +37,8 @@
 import threading
 from datetime import timedelta
 
+import _hashlib
+
 from elasticapm.conf.constants import BASE_SANITIZE_FIELD_NAMES, TRACE_CONTINUATION_STRATEGY
 from elasticapm.utils import compat, starmatch_to_regex
 from elasticapm.utils.logging import get_logger
@@ -373,6 +375,30 @@ def __call__(self, value, field_name):
         return value
 
 
+def _in_fips_mode():
+    try:
+        return _hashlib.get_fips_mode() == 1
+    except AttributeError:
+        # versions older of Python3.9 does not have the helper
+        return False
+
+
+class SupportedValueInFipsModeValidator(object):
+    """If FIPS mode is enabled only supported_value is accepted"""
+
+    def __init__(self, supported_value) -> None:
+        self.supported_value = supported_value
+
+    def __call__(self, value, field_name):
+        if _in_fips_mode():
+            if value != self.supported_value:
+                raise ConfigurationError(
+                    "{}={} must be set to {} if FIPS mode is enabled".format(field_name, value, self.supported_value),
+                    field_name,
+                )
+        return value
+
+
 class EnumerationValidator(object):
     """
     Validator which ensures that a given config value is chosen from a list
@@ -579,7 +605,9 @@ class Config(_ConfigBase):
     server_url = _ConfigValue("SERVER_URL", default="http://127.0.0.1:8200", required=True)
     server_cert = _ConfigValue("SERVER_CERT", validators=[FileIsReadableValidator()])
     server_ca_cert_file = _ConfigValue("SERVER_CA_CERT_FILE", validators=[FileIsReadableValidator()])
-    verify_server_cert = _BoolConfigValue("VERIFY_SERVER_CERT", default=True)
+    verify_server_cert = _BoolConfigValue(
+        "VERIFY_SERVER_CERT", default=True, validators=[SupportedValueInFipsModeValidator(supported_value=True)]
+    )
     use_certifi = _BoolConfigValue("USE_CERTIFI", default=True)
     include_paths = _ListConfigValue("INCLUDE_PATHS")
     exclude_paths = _ListConfigValue("EXCLUDE_PATHS", default=compat.get_default_library_patters())

tests/config/tests.py

@@ -39,6 +39,7 @@
 import mock
 import pytest
 
+import elasticapm.conf
 from elasticapm.conf import (
     Config,
     ConfigurationError,
@@ -47,6 +48,7 @@
     FileIsReadableValidator,
     PrecisionValidator,
     RegexValidator,
+    SupportedValueInFipsModeValidator,
     UnitValidator,
     VersionedConfig,
     _BoolConfigValue,
@@ -490,3 +492,31 @@ def test_exclude_range_validator_not_in_range():
     with pytest.raises(ConfigurationError) as e:
         validator(10, "field")
     assert "cannot be in range" in e.value.args[0]
+
+
+def test_supported_value_in_fips_mode_validator_in_fips_mode_with_invalid_value(monkeypatch):
+    monkeypatch.setattr(elasticapm.conf, "_in_fips_mode", lambda: True)
+    exception_message = "verify_server_cert=False must be set to True if FIPS mode is enabled"
+    validator = SupportedValueInFipsModeValidator(supported_value=True)
+    with pytest.raises(ConfigurationError) as e:
+        validator(False, "verify_server_cert")
+    assert exception_message == e.value.args[0]
+
+    config = Config({"VERIFY_SERVER_CERT": False})
+    assert config.errors["verify_server_cert"] == exception_message
+
+
+def test_supported_value_in_fips_mode_validator_in_fips_mode_with_valid_value(monkeypatch):
+    monkeypatch.setattr(elasticapm.conf, "_in_fips_mode", lambda: True)
+    validator = SupportedValueInFipsModeValidator(supported_value=True)
+    assert validator(True, "verify_server_cert") == True
+    config = Config({"VERIFY_SERVER_CERT": True})
+    assert config.verify_server_cert == True
+    assert "verify_server_cert" not in config.errors
+
+
+def test_supported_value_in_fips_mode_validator_not_in_fips_mode(monkeypatch):
+    monkeypatch.setattr(elasticapm.conf, "_in_fips_mode", lambda: False)
+    validator = SupportedValueInFipsModeValidator(supported_value=True)
+    assert validator(True, "field") == True
+    assert validator(False, "field") == False