Add a diagnostic for TLS certificate files (#15470)

* Diagnose TLS setup

* Add tests for diagnoses

mongo/datadog_checks/mongo/mongo.py

@@ -75,6 +75,8 @@ def __init__(self, name, init_config, instances=None):
         self._api_client = None
         self._mongo_version = None
 
+        self.diagnosis.register(self._diagnose_tls)
+
     @cached_property
     def api_client(self):
         # This needs to be a property for our unit test mocks to work.
@@ -247,3 +249,28 @@ def _get_db_names(self, api, deployment, tags):
                         )
         self.log.debug("List of databases to check: %s", dbnames)
         return dbnames
+
+    def _diagnose_tls(self):
+        # Check TLS config. Specifically, we might want to check that if `tls` is
+        # enabled (either explicitly or implicitly), the provided
+        # tls_certificate_key_file and tls_ca_file actually exist on the file system.
+        if "tls_certificate_key_file" in self.instance:
+            self._diagnose_readable('tls', self.instance["tls_certificate_key_file"], "tls_certificate_key_file")
+        if "tls_ca_file" in self.instance:
+            self._diagnose_readable('tls', self.instance["tls_ca_file"], "tls_ca_file")
+
+    def _diagnose_readable(self, name, path, option_name):
+        try:
+            open(path).close()
+        except FileNotFoundError:
+            self.diagnosis.fail(name, f"file `{path}` provided in the `{option_name}` option does not exist")
+        except OSError as exc:
+            self.diagnosis.fail(
+                name,
+                f"file `{path}` provided as the `{option_name}` option could not be opened: {exc.strerror}",
+            )
+        else:
+            self.diagnosis.success(
+                name,
+                f"file `{path}` provided as the `{option_name}` exists and is readable",
+            )

mongo/tests/test_diagnose.py

@@ -0,0 +1,51 @@
+# (C) Datadog, Inc. 2023-present
+# All rights reserved
+# Licensed under a 3-clause BSD style license (see LICENSE)
+import json
+from operator import itemgetter
+
+from datadog_checks.base.utils.diagnose import Diagnosis
+
+
+def test_no_diagnosis_when_no_certificates_are_specified(instance, check):
+    mongo_check = check(instance)
+    diagnoses = json.loads(mongo_check.get_diagnoses())
+    assert len(diagnoses) == 0
+
+
+def test_certificate_files_success(instance, check, tmp_path):
+    certificate_key_path = tmp_path / 'client.pem'
+    ca_path = tmp_path / 'ca.pem'
+
+    # Create dummy certificate files
+    certificate_key_path.touch()
+    ca_path.touch()
+
+    instance['tls_certificate_key_file'] = str(certificate_key_path)
+    instance['tls_ca_file'] = str(ca_path)
+
+    mongo_check = check(instance)
+    diagnoses = sorted(json.loads(mongo_check.get_diagnoses()), key=itemgetter('diagnosis'))
+    assert len(diagnoses) == 2
+    assert diagnoses[0]['result'] == Diagnosis.DIAGNOSIS_SUCCESS
+    assert str(ca_path) in diagnoses[0]['diagnosis']
+    assert diagnoses[1]['result'] == Diagnosis.DIAGNOSIS_SUCCESS
+    assert str(certificate_key_path) in diagnoses[1]['diagnosis']
+
+
+def test_certificate_files_failure_not_exist(instance, check, tmp_path):
+    certificate_key_path = tmp_path / 'client.pem'
+    ca_path = tmp_path / 'ca.pem'
+
+    instance['tls_certificate_key_file'] = str(certificate_key_path)
+    instance['tls_ca_file'] = str(ca_path)
+
+    mongo_check = check(instance)
+    diagnoses = sorted(json.loads(mongo_check.get_diagnoses()), key=itemgetter('diagnosis'))
+    assert len(diagnoses) == 2
+    assert diagnoses[0]['result'] == Diagnosis.DIAGNOSIS_FAIL
+    assert str(ca_path) in diagnoses[0]['diagnosis']
+    assert "does not exist" in diagnoses[0]['diagnosis']
+    assert diagnoses[1]['result'] == Diagnosis.DIAGNOSIS_FAIL
+    assert str(certificate_key_path) in diagnoses[1]['diagnosis']
+    assert "does not exist" in diagnoses[1]['diagnosis']