Add network policies

CONTRIBUTING.md

@@ -38,6 +38,8 @@ tox -e integration   # integration tests
 tox                  # runs 'fmt', 'lint', and 'unit' environments
 ```
 
+Running the integration tests requires that you have a valid kubeconfig file at `~/.kube/config`.
+
 ## Build charm
 
 Build the charm in this git repository using:

lib/charms/kratos/v0/kubernetes_network_policies.py

@@ -0,0 +1,160 @@
+#!/usr/bin/env python3
+# Copyright 2023 Canonical Ltd.
+# See LICENSE file for licensing details.
+
+"""Interface library for creating network policies.
+This library provides a Python API for creating kubernetes network policies.
+## Getting Started
+To get started using the library, you need to fetch the library using `charmcraft`.
+```shell
+cd some-charm
+charmcraft fetch-lib charms.kratos.v0.kubernetes_network_policies
+```
+Then, to initialise the library:
+```python
+from charms.kratos.v0.kubernetes_network_policies import (
+    K8sNetworkPoliciesHandler,
+    NetworkPoliciesHandlerError,
+    PortDefinition,
+)
+Class SomeCharm(CharmBase):
+    def __init__(self, *args):
+        self.network_policy_handler = K8sNetworkPoliciesHandler(self)
+
+    def some_event_function():
+        policies = [(PortDefinition("admin"), [self.admin_ingress_relation]), (PortDefinition(8080), [])]
+        self.network_policy_handler.apply_ingress_policy(policies)
+```
+
+The function in this example will only allow traffic to the charm pod to the "admin" port from the app on the
+other side of the `admin_ingress_relation` and all traffic to the "8080" port. Ingress traffic to all other ports
+will be denied.
+"""
+
+import logging
+from dataclasses import dataclass
+from typing import List, Optional, Tuple, Union
+
+from lightkube import ApiError, Client
+from lightkube.models.meta_v1 import LabelSelector, ObjectMeta
+from lightkube.models.networking_v1 import (
+    NetworkPolicyEgressRule,
+    NetworkPolicyIngressRule,
+    NetworkPolicyPeer,
+    NetworkPolicyPort,
+    NetworkPolicySpec,
+)
+from lightkube.resources.networking_v1 import NetworkPolicy
+from ops.charm import CharmBase
+from ops.model import Relation
+
+
+# The unique Charmhub library identifier, never change it
+LIBID = "f0a1c7a9bc084be09b1052810651b7ed"
+
+# Increment this major API version when introducing breaking changes
+LIBAPI = 0
+
+# Increment this PATCH version before using `charmcraft publish-lib` or reset
+# to 0 if you are raising the major API version
+LIBPATCH = 1
+
+logger = logging.getLogger(__name__)
+
+
+class NetworkPoliciesHandlerError(Exception):
+    """Applying the network policies failed."""
+
+
+Port = Union[str, int]
+IngressPolicyDefinition = Tuple[Port, List[Relation]]
+
+
+class KubernetesNetworkPoliciesHandler:
+    """A helper class for managing kubernetes network policies."""
+
+    def __init__(self, charm: CharmBase) -> None:
+        self._charm = charm
+        self.client = Client(field_manager=charm.app.name, namespace=charm.model.name)
+
+    @property
+    def policy_name(self) -> str:
+        """The default policy name that will be created."""
+        return f"{self._charm.app.name}-network-policy"
+
+    def apply_ingress_policy(
+        self, policies: IngressPolicyDefinition, name: Optional[str] = None
+    ) -> None:
+        """Apply an ingress network policy about a related application.
+
+        Policies can be defined for multiple ports at once to allow ingress traffic
+        from related applications
+
+        If no policies are defined then all ingress traffic will be denied.
+
+        Example usage:
+
+            policies = [("admin", [admin_ingress_relation]), (8080, [public_ingress_relation])]
+            network_policy_handler.apply_ingress_policy(policies)
+        """
+        if not name:
+            name = self.policy_name
+
+        ingress = []
+        for port, relations in policies:
+            selectors = [
+                NetworkPolicyPeer(
+                    podSelector=LabelSelector(
+                        matchLabels={"app.kubernetes.io/name": relation.app.name}
+                    )
+                )
+                for relation in relations
+                if relation and relation.app
+            ]
+            ingress.append(
+                NetworkPolicyIngressRule(
+                    from_=selectors,
+                    ports=[NetworkPolicyPort(port=port, protocol="TCP")],
+                ),
+            )
+
+        policy = NetworkPolicy(
+            metadata=ObjectMeta(name=name),
+            spec=NetworkPolicySpec(
+                podSelector=LabelSelector(
+                    matchLabels={
+                        "app.kubernetes.io/name": self._charm.app.name,
+                    }
+                ),
+                policyTypes=["Ingress", "Egress"],
+                ingress=ingress,
+            ),
+        )
+
+        try:
+            self.client.apply(
+                policy,
+                namespace=self._charm.model.name,
+            )
+        except ApiError as e:
+            if e.status.code == 403:
+                msg = f"Kubernetes resources patch failed: `juju trust` this application. {e}"
+            else:
+                msg = f"Kubernetes resources patch failed: {e}"
+            logger.error(msg)
+            raise NetworkPoliciesHandlerError()
+
+    def delete_network_policy(self, name: Optional[str] = None) -> None:
+        """Delete a network policy rule."""
+        if not name:
+            name = self.policy_name
+
+        try:
+            self.client.delete(name, namespace=self._charm.model.name)
+        except ApiError as e:
+            if e.status.code == 403:
+                msg = f"Kubernetes resources patch failed: `juju trust` this application. {e}"
+            else:
+                msg = f"Kubernetes resources patch failed: {e}"
+            logger.error(msg)
+            raise NetworkPoliciesHandlerError()

src/charm.py

@@ -31,6 +31,10 @@
     LoginUITooManyRelatedAppsError,
 )
 from charms.kratos.v0.kratos_endpoints import KratosEndpointsProvider
+from charms.kratos.v0.kubernetes_network_policies import (
+    KubernetesNetworkPoliciesHandler,
+    NetworkPoliciesHandlerError,
+)
 from charms.kratos_external_idp_integrator.v0.kratos_external_provider import (
     ClientConfigChangedEvent,
     ExternalIdpRequirer,
@@ -49,6 +53,7 @@
     HookEvent,
     PebbleReadyEvent,
     RelationEvent,
+    RemoveEvent,
 )
 from ops.main import main
 from ops.model import ActiveStatus, BlockedStatus, MaintenanceStatus, ModelError, WaitingStatus
@@ -117,6 +122,7 @@ def __init__(self, *args: Any) -> None:
             port=KRATOS_PUBLIC_PORT,
             strip_prefix=True,
         )
+        self.network_policy_handler = KubernetesNetworkPoliciesHandler(self)
 
         self.database = DatabaseRequires(
             self,
@@ -139,6 +145,7 @@ def __init__(self, *args: Any) -> None:
 
         self.framework.observe(self.on.kratos_pebble_ready, self._on_pebble_ready)
         self.framework.observe(self.on.config_changed, self._on_config_changed)
+        self.framework.observe(self.on.remove, self._cleanup)
         self.framework.observe(
             self.endpoints_provider.on.ready, self._update_kratos_endpoints_relation_data
         )
@@ -151,9 +158,13 @@ def __init__(self, *args: Any) -> None:
         self.framework.observe(self.database.on.database_created, self._on_database_created)
         self.framework.observe(self.database.on.endpoints_changed, self._on_database_changed)
         self.framework.observe(self.admin_ingress.on.ready, self._on_admin_ingress_ready)
+        self.framework.observe(self.admin_ingress.on.ready, self._apply_network_policies)
         self.framework.observe(self.admin_ingress.on.revoked, self._on_ingress_revoked)
+        self.framework.observe(self.admin_ingress.on.revoked, self._apply_network_policies)
         self.framework.observe(self.public_ingress.on.ready, self._on_public_ingress_ready)
+        self.framework.observe(self.public_ingress.on.ready, self._apply_network_policies)
         self.framework.observe(self.public_ingress.on.revoked, self._on_ingress_revoked)
+        self.framework.observe(self.public_ingress.on.revoked, self._apply_network_policies)
         self.framework.observe(
             self.external_provider.on.client_config_changed, self._on_client_config_changed
         )
@@ -442,6 +453,9 @@ def _on_config_changed(self, event: ConfigChangedEvent) -> None:
         """Event Handler for config changed event."""
         self._handle_status_update_config(event)
 
+    def _cleanup(self, event: RemoveEvent) -> None:
+        self.network_policy_handler.delete_network_policy()
+
     def _update_kratos_endpoints_relation_data(self, event: RelationEvent) -> None:
         admin_endpoint = (
             self.admin_ingress.url
@@ -522,6 +536,22 @@ def _on_admin_ingress_ready(self, event: IngressPerAppReadyEvent) -> None:
         self._handle_status_update_config(event)
         self._update_kratos_endpoints_relation_data(event)
 
+    def _apply_network_policies(self, event: HookEvent) -> None:
+        if not self.unit.is_leader():
+            return
+
+        try:
+            self.network_policy_handler.apply_ingress_policy(
+                [
+                    (KRATOS_PUBLIC_PORT, [self.public_ingress.relation]),
+                    (KRATOS_ADMIN_PORT, [self.admin_ingress.relation]),
+                    (38812, []),
+                    (38813, []),
+                ]
+            )
+        except NetworkPoliciesHandlerError:
+            event.defer()
+
     def _on_public_ingress_ready(self, event: IngressPerAppReadyEvent) -> None:
         if self.unit.is_leader():
             logger.info("This app's public ingress URL: %s", event.url)

tests/integration/conftest.py

@@ -0,0 +1,14 @@
+#!/usr/bin/env python3
+# Copyright 2022 Canonical Ltd.
+
+import os
+
+import pytest
+from lightkube import Client, KubeConfig
+
+KUBECONFIG = os.environ.get("TESTING_KUBECONFIG", "~/.kube/config")
+
+
+@pytest.fixture(scope="module")
+def client() -> Client:
+    return Client(config=KubeConfig.from_file(KUBECONFIG))

tests/integration/test_charm.py

@@ -9,6 +9,8 @@
 import pytest
 import requests
 import yaml
+from lightkube import Client
+from lightkube.resources.networking_v1 import NetworkPolicy
 from pytest_operator.plugin import OpsTest
 
 logger = logging.getLogger(__name__)
@@ -49,29 +51,14 @@ async def test_build_and_deploy(ops_test: OpsTest) -> None:
 
     Assert on the unit status before any relations/configurations take place.
     """
-    await ops_test.model.deploy(
-        POSTGRES,
-        channel="latest/edge",
-        trust=True,
-    )
     charm = await ops_test.build_charm(".")
     resources = {"oci-image": METADATA["resources"]["oci-image"]["upstream-source"]}
     await ops_test.model.deploy(
         charm, resources=resources, application_name=APP_NAME, trust=True, series="jammy"
     )
-    await ops_test.model.add_relation(APP_NAME, POSTGRES)
-
-    async with ops_test.fast_forward():
-        await ops_test.model.wait_for_idle(
-            apps=[APP_NAME, POSTGRES],
-            status="active",
-            raise_on_blocked=True,
-            timeout=1000,
-        )
-        assert ops_test.model.applications[APP_NAME].units[0].workload_status == "active"
 
 
-async def test_ingress_relation(ops_test: OpsTest) -> None:
+async def test_ingress_relation(ops_test: OpsTest, client: Client) -> None:
     await ops_test.model.deploy(
         TRAEFIK,
         application_name=TRAEFIK_PUBLIC_APP,
@@ -94,6 +81,27 @@ async def test_ingress_relation(ops_test: OpsTest) -> None:
         timeout=1000,
     )
 
+    # Validate network policies are created when ingress is provided
+    policy = client.get(NetworkPolicy, "kratos-network-policy", namespace=ops_test.model.name)
+    assert policy
+
+
+async def test_postgres_relation(ops_test: OpsTest) -> None:
+    await ops_test.model.deploy(
+        POSTGRES,
+        channel="14/stable",
+        trust=True,
+    )
+    await ops_test.model.add_relation(APP_NAME, POSTGRES)
+
+    async with ops_test.fast_forward():
+        await ops_test.model.wait_for_idle(
+            apps=[APP_NAME, POSTGRES],
+            status="active",
+            timeout=1000,
+        )
+        assert ops_test.model.applications[APP_NAME].units[0].workload_status == "active"
+
 
 async def test_has_public_ingress(ops_test: OpsTest) -> None:
     # Get the traefik address and try to reach kratos

tests/unit/conftest.py

@@ -15,6 +15,11 @@
 from kratos import KratosAPI
 
 
+@pytest.fixture(autouse=True)
+def lk_client(mocker):
+    return mocker.patch("charm.KubernetesNetworkPoliciesHandler")
+
+
 @pytest.fixture()
 def harness(mocked_kubernetes_service_patcher: MagicMock) -> Harness:
     harness = Harness(KratosCharm)