feat: Allow exposing Service objects with annotations (#244)

app/handlers/__init__.py

@@ -2,3 +2,4 @@
 from .handlers_connectors import *
 from .handlers_resource import *
 from .handlers_resource_access import *
+from .handlers_services import *

app/handlers/handlers_services.py

@@ -0,0 +1,101 @@
+import kopf
+import kubernetes
+
+from app.handlers import success
+
+
+def k8s_get_twingate_resource(
+    namespace: str, name: str, kapi: kubernetes.client.CustomObjectsApi | None = None
+) -> dict | None:
+    kapi = kapi or kubernetes.client.CustomObjectsApi()
+    try:
+        return kapi.get_namespaced_custom_object(
+            "twingate.com", "v1beta", namespace, "twingateresources", name
+        )
+    except kubernetes.client.exceptions.ApiException as ex:
+        if ex.status == 404:
+            return None
+        raise
+
+
+ALLOWED_EXTRA_ANNOTATIONS = [
+    "alias",
+    "isBrowserShortcutEnabled",
+    "securityPolicyId",
+    "isVisible",
+]
+
+
+def service_to_twingate_resource(service_body, namespace) -> dict:
+    meta = service_body.metadata
+    spec = service_body.spec
+    service_name = service_body.meta.name
+    resource_object_name = f"{service_name}-resource"
+
+    result: dict = {
+        "apiVersion": "twingate.com/v1beta",
+        "kind": "TwingateResource",
+        "metadata": {
+            "name": resource_object_name,
+        },
+        "spec": {
+            "name": resource_object_name,
+            "address": f"{service_name}.{namespace}.svc.cluster.local",
+        },
+    }
+
+    for key in ALLOWED_EXTRA_ANNOTATIONS:
+        if value := meta.annotations.get(f"twingate.com/resource-{key}"):
+            result["spec"][key] = value
+
+    if service_ports := spec.get("ports", []):
+        protocols: dict = {
+            "allowIcmp": False,
+            "tcp": {"policy": "RESTRICTED", "ports": []},
+            "udp": {"policy": "RESTRICTED", "ports": []},
+        }
+        for port_obj in service_ports:
+            port = port_obj["port"]
+            if port_obj["protocol"] == "TCP":
+                protocols["tcp"]["ports"].append({"start": port, "end": port})
+            elif port_obj["protocol"] == "UDP":
+                protocols["udp"]["ports"].append({"start": port, "end": port})
+
+        result["spec"]["protocols"] = protocols
+
+    return result
+
+
+@kopf.on.resume("service", annotations={"twingate.com/resource": "true"})
+@kopf.on.create("service", annotations={"twingate.com/resource": "true"})
+@kopf.on.update("service", annotations={"twingate.com/resource": "true"})
+def twingate_service_create(body, spec, namespace, meta, logger, **_):
+    logger.info("twingate_service_create: %s", spec)
+
+    resource_subobject = service_to_twingate_resource(body, namespace)
+    kopf.adopt(resource_subobject)
+
+    resource_object_name = resource_subobject["metadata"]["name"]
+
+    kapi = kubernetes.client.CustomObjectsApi()
+    if existing_resource_object := k8s_get_twingate_resource(
+        namespace, resource_object_name, kapi
+    ):
+        logger.info("TwingateResource already exists: %s", existing_resource_object)
+        kapi.patch_namespaced_custom_object(
+            "twingate.com",
+            "v1beta",
+            namespace,
+            "twingateresources",
+            resource_object_name,
+            resource_subobject,
+        )
+    else:
+        api_response = kapi.create_namespaced_custom_object(
+            "twingate.com", "v1beta", namespace, "twingateresources", resource_subobject
+        )
+        logger.info("create_namespaced_custom_object response: %s", api_response)
+
+    return success(
+        message=f"Created TwingateResource {resource_subobject['spec']['name']}"
+    )

app/handlers/tests/test_handlers_services.py

@@ -0,0 +1,164 @@
+from unittest.mock import MagicMock, patch
+
+import kopf
+import kubernetes
+import pytest
+import yaml  # type: ignore
+
+from app.handlers.handlers_services import (
+    ALLOWED_EXTRA_ANNOTATIONS,
+    k8s_get_twingate_resource,
+    service_to_twingate_resource,
+    twingate_service_create,
+)
+
+# Ignore the fact we use _cogs here
+
+
+@pytest.fixture()
+def example_service_body():
+    yaml_str = """
+    apiVersion: v1
+    kind: Service
+    metadata:
+      name: my-service
+      annotations:
+        twingate.com/resource: "true"
+        twingate.com/resource-alias: "myapp.internal"
+    spec:
+      selector:
+        app.kubernetes.io/name: MyApp
+      ports:
+        - name: http
+          protocol: TCP
+          port: 80
+          targetPort: 9376
+        - name: https
+          protocol: TCP
+          port: 443
+          targetPort: 9377
+        - protocol: UDP
+          port: 22
+          targetPort: 9376
+          name: ssh
+    """
+    return kopf._cogs.structs.bodies.Body(  # noqa: SLF001
+        yaml.safe_load(yaml_str)
+    )
+
+
+@pytest.fixture()
+def k8s_customobjects_client_mock():
+    client_mock = MagicMock()
+    with patch("kubernetes.client.CustomObjectsApi") as k8sclient_mock:
+        k8sclient_mock.return_value = client_mock
+        yield client_mock
+
+
+class TestServiceToTwingateResource:
+    @pytest.mark.parametrize("annotation_name", ["", *ALLOWED_EXTRA_ANNOTATIONS])
+    def test_with_extra_annotation(self, example_service_body, annotation_name):
+        expected = {
+            "apiVersion": "twingate.com/v1beta",
+            "kind": "TwingateResource",
+            "metadata": {
+                "name": "my-service-resource",
+            },
+            "spec": {
+                "name": "my-service-resource",
+                "address": "my-service.default.svc.cluster.local",
+                "alias": "myapp.internal",
+                "protocols": {
+                    "allowIcmp": False,
+                    "tcp": {
+                        "policy": "RESTRICTED",
+                        "ports": [{"start": 80, "end": 80}, {"start": 443, "end": 443}],
+                    },
+                    "udp": {
+                        "policy": "RESTRICTED",
+                        "ports": [{"start": 22, "end": 22}],
+                    },
+                },
+            },
+        }
+
+        if annotation_name:
+            example_service_body.metadata["annotations"][
+                f"twingate.com/resource-{annotation_name}"
+            ] = f"{annotation_name} value"
+
+            expected["spec"][annotation_name] = f"{annotation_name} value"
+
+        result = service_to_twingate_resource(example_service_body, "default")
+        assert result == expected
+
+
+class TestK8sGetTwingateResource:
+    def test_handles_404_returns_none(
+        self,
+        k8s_customobjects_client_mock,
+    ):
+        k8s_customobjects_client_mock.get_namespaced_custom_object.side_effect = (
+            kubernetes.client.exceptions.ApiException(status=404)
+        )
+        assert k8s_get_twingate_resource("default", "test") is None
+
+    def test_reraises_non_404_exceptions(
+        self,
+        k8s_customobjects_client_mock,
+    ):
+        k8s_customobjects_client_mock.get_namespaced_custom_object.side_effect = (
+            kubernetes.client.exceptions.ApiException(status=500)
+        )
+        with pytest.raises(kubernetes.client.exceptions.ApiException):
+            k8s_get_twingate_resource("default", "test")
+
+
+class TestTwingateServiceCreate:
+    def test_create_service_triggers_creation_of_twingate_resource(
+        self, example_service_body, kopf_handler_runner, k8s_customobjects_client_mock
+    ):
+        k8s_customobjects_client_mock.get_namespaced_custom_object.return_value = None
+
+        twingate_service_create(
+            example_service_body,
+            example_service_body.spec,
+            "default",
+            example_service_body.metadata,
+            MagicMock(),
+        )
+
+        k8s_customobjects_client_mock.patch_namespaced_custom_object.assert_not_called()
+        k8s_customobjects_client_mock.create_namespaced_custom_object.assert_called_once_with(
+            "twingate.com",
+            "v1beta",
+            "default",
+            "twingateresources",
+            service_to_twingate_resource(example_service_body, "default"),
+        )
+
+    def test_update_service_propogates_changes_to_twingate_resource(
+        self, example_service_body, kopf_handler_runner, k8s_customobjects_client_mock
+    ):
+        k8s_customobjects_client_mock.get_namespaced_custom_object.return_value = {
+            "metadata": {"name": "my-service-resource"},
+            "spec": {"address": "my-service.default.svc.cluster.local"},
+        }
+
+        twingate_service_create(
+            example_service_body,
+            example_service_body.spec,
+            "default",
+            example_service_body.metadata,
+            MagicMock(),
+        )
+
+        k8s_customobjects_client_mock.patch_namespaced_custom_object.assert_called_once_with(
+            "twingate.com",
+            "v1beta",
+            "default",
+            "twingateresources",
+            "my-service-resource",
+            service_to_twingate_resource(example_service_body, "default"),
+        )
+        k8s_customobjects_client_mock.create_namespaced_custom_object.assert_not_called()

examples/service.yaml

@@ -0,0 +1,19 @@
+apiVersion: v1
+kind: Service
+metadata:
+  name: my-service
+  annotations:
+    twingate.com/resource: "true"
+    twingate.com/resource-alias: "myapp.internal"
+spec:
+  selector:
+    app.kubernetes.io/name: MyApp
+  ports:
+    - protocol: TCP
+      port: 80
+      targetPort: 9376
+      name: first
+    - protocol: UDP
+      port: 22
+      targetPort: 9376
+      name: second

tests_integration/test_service_flows.py

@@ -0,0 +1,105 @@
+import time
+from subprocess import CalledProcessError
+from unittest.mock import ANY
+
+import pytest
+from kopf.testing import KopfRunner
+
+from tests_integration.utils import (
+    kubectl_create,
+    kubectl_delete,
+    kubectl_get,
+    kubectl_patch,
+)
+
+
+def test_service_flows(kopf_runner_args, kopf_settings, ci_run_number):
+    service_name = f"my-service-{ci_run_number}"
+    resource_name = f"{service_name}-resource"
+    SERVICE_OBJ = f"""
+        apiVersion: v1
+        kind: Service
+        metadata:
+          name: {service_name}
+          annotations:
+            twingate.com/resource: "true"
+            twingate.com/resource-alias: "myapp.internal"
+        spec:
+          selector:
+            app.kubernetes.io/name: MyApp
+          ports:
+            - name: http
+              protocol: TCP
+              port: 80
+              targetPort: 9376
+            - protocol: UDP
+              port: 22
+              targetPort: 9376
+              name: ssh
+    """
+
+    with KopfRunner(kopf_runner_args, settings=kopf_settings) as runner:
+        kubectl_create(SERVICE_OBJ)
+        time.sleep(2)
+
+        kubectl_get("service", service_name)
+        tgr = kubectl_get("twingateresource", resource_name)
+
+        assert tgr["spec"] == {
+            "address": f"{service_name}.default.svc.cluster.local",
+            "alias": "myapp.internal",
+            "id": ANY,
+            "isBrowserShortcutEnabled": False,
+            "isVisible": True,
+            "name": f"{service_name}-resource",
+            "protocols": {
+                "allowIcmp": False,
+                "tcp": {"policy": "RESTRICTED", "ports": [{"end": 80, "start": 80}]},
+                "udp": {"policy": "RESTRICTED", "ports": [{"end": 22, "start": 22}]},
+            },
+        }
+
+        # Test patching the service updates the resource
+        kubectl_patch(
+            f"svc/{service_name}",
+            [
+                {
+                    "op": "add",
+                    "path": "/spec/ports/-",
+                    "value": {
+                        "protocol": "TCP",
+                        "port": 443,
+                        "targetPort": 9377,
+                        "name": "https",
+                    },
+                }
+            ],
+            "json",
+        )
+        time.sleep(2)
+        tgr = kubectl_get("twingateresource", resource_name)
+        assert tgr["spec"] == {
+            "address": f"{service_name}.default.svc.cluster.local",
+            "alias": "myapp.internal",
+            "id": ANY,
+            "isBrowserShortcutEnabled": False,
+            "isVisible": True,
+            "name": f"{service_name}-resource",
+            "protocols": {
+                "allowIcmp": False,
+                "tcp": {
+                    "policy": "RESTRICTED",
+                    "ports": [{"end": 80, "start": 80}, {"end": 443, "start": 443}],
+                },
+                "udp": {"policy": "RESTRICTED", "ports": [{"end": 22, "start": 22}]},
+            },
+        }
+
+        # Test deleting the service deletes the resource
+        kubectl_delete(f"service/{service_name}")
+        time.sleep(5)
+        with pytest.raises(CalledProcessError):
+            kubectl_get("twingateresource", resource_name)
+
+    assert runner.exception is None
+    assert runner.exit_code == 0