[DPE-3926] Enforce zookeeper client interface (#106)

.safety-policy.yml

@@ -0,0 +1,5 @@
+security:
+    ignore-vulnerabilities:
+        67599:
+          reason: CVE-2018-20225 is disputed, and present in all pip versions by design.
+          # expires: "never"

pyproject.toml

@@ -70,7 +70,7 @@ optional = true
 
 [tool.poetry.group.security.dependencies]
 bandit = "^1.7.4"
-safety = "^2.3.5"
+safety = "^3.0.0"
 
 
 [tool.poetry.group.integration]

src/core/models.py

@@ -33,9 +33,13 @@ def __init__(
     ):
         self.relation = relation
         self.data_interface = data_interface
-        self.component = component
+        self.component = (
+            component  # FIXME: remove, and use _fetch_my_relation_data defaults wheren needed
+        )
         self.substrate = substrate
-        self.relation_data = self.data_interface.as_dict(self.relation.id) if self.relation else {}
+        self.relation_data = (
+            self.data_interface.as_dict(self.relation.id) if self.relation else {}
+        )  # FIXME: mappingproxytype?
 
     def __bool__(self) -> bool:
         """Boolean evaluation based on the existence of self.relation."""
@@ -280,6 +284,19 @@ def endpoints(self) -> str:
             or ""
         )
 
+    @property
+    def database(self) -> str:
+        """Path allocated for Kafka on ZooKeeper."""
+        if not self.relation:
+            return ""
+
+        return (
+            self.data_interface.fetch_relation_field(
+                relation_id=self.relation.id, field="database"
+            )
+            or self.chroot
+        )
+
     @property
     def chroot(self) -> str:
         """Path allocated for Kafka on ZooKeeper."""
@@ -316,8 +333,8 @@ def tls(self) -> bool:
     @property
     def connect(self) -> str:
         """Full connection string of sorted uris."""
-        sorted_uris = sorted(self.uris.replace(self.chroot, "").split(","))
-        sorted_uris[-1] = sorted_uris[-1] + self.chroot
+        sorted_uris = sorted(self.uris.replace(self.database, "").split(","))
+        sorted_uris[-1] = sorted_uris[-1] + self.database
         return ",".join(sorted_uris)
 
     @property
@@ -328,7 +345,7 @@ def zookeeper_connected(self) -> bool:
             True if ZooKeeper is currently related with sufficient relation data
                 for a broker to connect with. Otherwise False
         """
-        if not all([self.username, self.password, self.endpoints, self.chroot, self.uris]):
+        if not all([self.username, self.password, self.endpoints, self.database, self.uris]):
             return False
 
         return True
@@ -352,7 +369,7 @@ def broker_active(self) -> bool:
         """Checks if broker id is recognised as active by ZooKeeper."""
         broker_id = self.data_interface.local_unit.name.split("/")[1]
         hosts = self.endpoints.split(",")
-        path = f"{self.chroot}/brokers/ids/"
+        path = f"{self.database}/brokers/ids/"
 
         zk = ZooKeeperManager(hosts=hosts, username=self.username, password=self.password)
         try:

src/events/zookeeper.py

@@ -38,7 +38,13 @@ def __init__(self, charm) -> None:
     def _on_zookeeper_created(self, _) -> None:
         """Handler for `zookeeper_relation_created` events."""
         if self.model.unit.is_leader():
-            self.charm.state.zookeeper.update({"chroot": "/" + self.model.app.name})
+            self.charm.state.zookeeper.update(
+                {
+                    "database": "/" + self.model.app.name,
+                    "requested-secrets": '["username","password","tls","tls-ca","uris"]',
+                    "chroot": "/" + self.model.app.name,
+                }
+            )
 
     def _on_zookeeper_changed(self, event: RelationChangedEvent) -> None:
         """Handler for `zookeeper_relation_created/joined/changed` events, ensuring internal users get created."""

tests/integration/helpers.py

@@ -432,7 +432,7 @@ def get_active_brokers(config: Dict) -> Set[str]:
     Returns:
         Set of active broker ids
     """
-    chroot = config.get("chroot", "")
+    chroot = config.get("database", config.get("chroot", ""))
     hosts = config.get("endpoints", "").split(",")
     username = config.get("username", "")
     password = config.get("password", "")

tests/integration/test_scaling.py

@@ -59,7 +59,7 @@ async def test_kafka_simple_scale_up(ops_test: OpsTest):
 
     active_brokers = get_active_brokers(config=kafka_zk_relation_data)
 
-    chroot = kafka_zk_relation_data.get("chroot", "")
+    chroot = kafka_zk_relation_data.get("database", kafka_zk_relation_data.get("chroot", ""))
     assert f"{chroot}/brokers/ids/0" in active_brokers
     assert f"{chroot}/brokers/ids/1" in active_brokers
     assert f"{chroot}/brokers/ids/2" in active_brokers
@@ -87,7 +87,7 @@ async def test_kafka_simple_scale_down(ops_test: OpsTest):
     kafka_zk_relation_data["endpoints"] = zookeeper_address
 
     active_brokers = get_active_brokers(config=kafka_zk_relation_data)
-    chroot = kafka_zk_relation_data.get("chroot", "")
+    chroot = kafka_zk_relation_data.get("database", kafka_zk_relation_data.get("chroot", ""))
     assert f"{chroot}/brokers/ids/0" in active_brokers
     assert f"{chroot}/brokers/ids/1" in active_brokers
     assert f"{chroot}/brokers/ids/2" not in active_brokers

tests/integration/test_tls.py

@@ -190,7 +190,7 @@ async def test_kafka_tls_scaling(ops_test: OpsTest):
 
     active_brokers = get_active_brokers(config=kafka_zk_relation_data)
 
-    chroot = kafka_zk_relation_data.get("chroot", "")
+    chroot = kafka_zk_relation_data.get("database", kafka_zk_relation_data.get("chroot", ""))
     assert f"{chroot}/brokers/ids/0" in active_brokers
     assert f"{chroot}/brokers/ids/1" in active_brokers
     assert f"{chroot}/brokers/ids/2" in active_brokers

tests/unit/conftest.py

@@ -14,6 +14,7 @@ def zk_data() -> dict[str, str]:
         "username": "glorfindel",
         "password": "mellon",
         "endpoints": "10.10.10.10",
+        "database": "/kafka",
         "chroot": "/kafka",
         "uris": "10.10.10.10:2181",
         "tls": "disabled",

tests/unit/test_charm.py

@@ -501,9 +501,8 @@ def test_zookeeper_joined_sets_chroot(harness: Harness):
     zk_rel_id = harness.add_relation(ZK, ZK)
     harness.add_relation_unit(zk_rel_id, f"{ZK}/0")
 
-    assert CHARM_KEY in harness.charm.model.relations[ZK][0].data[harness.charm.app].get(
-        "chroot", ""
-    )
+    rel = harness.charm.model.relations[ZK][0].data[harness.charm.app]
+    assert CHARM_KEY in rel.get("database", rel.get("chroot", ""))
 
 
 def test_zookeeper_broken_stops_service_and_removes_meta_properties(harness: Harness):

tests/unit/test_config.py

@@ -83,6 +83,7 @@ def test_log_dirs_in_server_properties(harness: Harness):
         zk_relation_id,
         harness.charm.app.name,
         {
+            "database": "/kafka",
             "chroot": "/kafka",
             "username": "moria",
             "password": "mellon",
@@ -119,6 +120,7 @@ def test_listeners_in_server_properties(harness: Harness):
         zk_relation_id,
         harness.charm.app.name,
         {
+            "database": "/kafka",
             "chroot": "/kafka",
             "username": "moria",
             "password": "mellon",
@@ -162,6 +164,7 @@ def test_ssl_listeners_in_server_properties(harness: Harness):
         zk_relation_id,
         harness.charm.app.name,
         {
+            "database": "/kafka",
             "chroot": "/kafka",
             "username": "moria",
             "password": "mellon",
@@ -206,6 +209,7 @@ def test_zookeeper_config_succeeds_fails_config(harness: Harness):
         zk_relation_id,
         harness.charm.app.name,
         {
+            "database": "/kafka",
             "chroot": "/kafka",
             "username": "moria",
             "endpoints": "1.1.1.1,2.2.2.2",
@@ -223,6 +227,7 @@ def test_zookeeper_config_succeeds_valid_config(harness: Harness):
         zk_relation_id,
         harness.charm.app.name,
         {
+            "database": "/kafka",
             "chroot": "/kafka",
             "username": "moria",
             "password": "mellon",
@@ -336,6 +341,7 @@ def test_ssl_principal_mapping_rules(harness: Harness):
         zk_relation_id,
         harness.charm.app.name,
         {
+            "database": "/kafka",
             "chroot": "/kafka",
             "username": "moria",
             "password": "mellon",
@@ -376,6 +382,7 @@ def test_auth_properties(harness: Harness):
         zk_relation_id,
         harness.charm.app.name,
         {
+            "database": "/kafka",
             "chroot": "/kafka",
             "username": "moria",
             "password": "mellon",
@@ -400,6 +407,7 @@ def test_rack_properties(harness: Harness):
         zk_relation_id,
         harness.charm.app.name,
         {
+            "database": "/kafka",
             "chroot": "/kafka",
             "username": "moria",
             "password": "mellon",
@@ -427,6 +435,7 @@ def test_inter_broker_protocol_version(harness: Harness):
         zk_relation_id,
         harness.charm.app.name,
         {
+            "database": "/kafka",
             "chroot": "/kafka",
             "username": "moria",
             "password": "mellon",

tests/unit/test_tls.py

@@ -41,6 +41,7 @@ def harness():
         zk_relation_id,
         harness.charm.app.name,
         {
+            "database": "/kafka",
             "chroot": "/kafka",
             "username": "moria",
             "password": "mellon",