Working version

This is fairly complex, will probably have to refactor it

src/charm.py

@@ -54,7 +54,11 @@
 from ops.model import ActiveStatus, BlockedStatus, MaintenanceStatus, ModelError, WaitingStatus
 from ops.pebble import Error, ExecError, Layer
 
-from k8s_network_policies import K8sNetworkPoliciesHandler, NetworkPoliciesHandlerError
+from k8s_network_policies import (
+    K8sNetworkPoliciesHandler,
+    NetworkPoliciesHandlerError,
+    PortDefinition,
+)
 from kratos import KratosAPI
 
 if TYPE_CHECKING:
@@ -535,8 +539,10 @@ def _apply_network_policies(self, event: HookEvent) -> None:
         try:
             self.network_policy_handler.apply_ingress_policy(
                 [
-                    ("admin", [self.admin_ingress.relation]),
-                    ("public", [self.public_ingress.relation]),
+                    (PortDefinition(1, KRATOS_PUBLIC_PORT - 1), ()),
+                    (PortDefinition(KRATOS_PUBLIC_PORT), [self.public_ingress.relation]),
+                    (PortDefinition(KRATOS_ADMIN_PORT), [self.admin_ingress.relation]),
+                    (PortDefinition(KRATOS_ADMIN_PORT + 1, 65535), ()),
                 ]
             )
         except NetworkPoliciesHandlerError:

src/k8s_network_policies.py

@@ -5,6 +5,7 @@
 """A helper class for managing kubernetes network policies."""
 
 import logging
+from dataclasses import dataclass
 from typing import List, Optional, Tuple, Union
 
 from lightkube import ApiError, Client
@@ -26,7 +27,22 @@ class NetworkPoliciesHandlerError(Exception):
     """Applying the network policies failed."""
 
 
-IngressPolicyDefinition = Tuple[Union[str, int], List[Relation]]
+@dataclass
+class PortDefinition:
+    """Network Policy port definition."""
+
+    port: Union[str, int]
+    end_port: Optional[int] = None
+    protocol: Optional[str] = "TCP"
+
+    def to_resource(self):
+        """Convert class to NetworkPolicyPort."""
+        if not self.end_port:
+            return NetworkPolicyPort(port=self.port, protocol=self.protocol)
+        return NetworkPolicyPort(port=self.port, endPort=self.end_port, protocol=self.protocol)
+
+
+IngressPolicyDefinition = Tuple[PortDefinition, List[Relation]]
 
 
 class K8sNetworkPoliciesHandler:
@@ -73,7 +89,7 @@ def apply_ingress_policy(
             ingress.append(
                 NetworkPolicyIngressRule(
                     from_=selectors,
-                    ports=[NetworkPolicyPort(port=port)],
+                    ports=[port.to_resource()],
                 ),
             )
 
@@ -86,9 +102,8 @@ def apply_ingress_policy(
                         "kubernetes.io/metadata.name": self._charm.model.name,
                     }
                 ),
-                policyTypes=["Ingress", "Egress"],
+                policyTypes=["Ingress"],
                 ingress=ingress,
-                egress=[{}]
             ),
         )
 

tests/integration/test_charm.py

@@ -97,7 +97,7 @@ async def test_ingress_relation(ops_test: OpsTest, client: Client) -> None:
     )
 
     # Validate network policies are created when ingress is provided
-    policy = client.get(NetworkPolicy, "kratos-network-policy")
+    policy = client.get(NetworkPolicy, "kratos-network-policy", namespace=ops_test.model.name)
     assert policy