feat(network): add security group binding to port to mcp tools (#87)

Author: platanus-kr

src/openstack_mcp_server/tools/network_tools.py

@@ -44,6 +44,8 @@ def register_tools(self, mcp: FastMCP):
         mcp.tool()(self.delete_port)
         mcp.tool()(self.get_port_allowed_address_pairs)
         mcp.tool()(self.set_port_binding)
+        mcp.tool()(self.add_security_group_to_port)
+        mcp.tool()(self.remove_security_group_from_port)
         mcp.tool()(self.get_floating_ips)
         mcp.tool()(self.create_floating_ip)
         mcp.tool()(self.delete_floating_ip)
@@ -515,6 +517,62 @@ def set_port_binding(
         updated = conn.network.update_port(port_id, **update_args)
         return self._convert_to_port_model(updated)
 
+    def add_security_group_to_port(
+        self,
+        port_id: str,
+        security_group_id: str,
+    ) -> Port:
+        """
+        Attach a Security Group to a Port.
+
+        Idempotent: if the security group is already attached, returns the current
+        port state without issuing an update.
+
+        :param port_id: Port ID
+        :param security_group_id: Security Group ID to attach
+        :return: Updated (or current) Port object
+        """
+        conn = get_openstack_conn()
+        current = conn.network.get_port(port_id)
+        current_group_ids: list[str] = list(current.security_group_ids or [])
+        if security_group_id in current_group_ids:
+            return self._convert_to_port_model(current)
+
+        updated_group_ids = current_group_ids + [security_group_id]
+        updated = conn.network.update_port(
+            port_id, security_groups=updated_group_ids
+        )
+        return self._convert_to_port_model(updated)
+
+    def remove_security_group_from_port(
+        self,
+        port_id: str,
+        security_group_id: str,
+    ) -> Port:
+        """
+        Detach a Security Group from a Port.
+
+        Idempotent: if the security group is not attached, returns the current
+        port state without issuing an update.
+
+        :param port_id: Port ID
+        :param security_group_id: Security Group ID to detach
+        :return: Updated (or current) Port object
+        """
+        conn = get_openstack_conn()
+        current = conn.network.get_port(port_id)
+        current_group_ids: list[str] = list(current.security_group_ids or [])
+        if security_group_id not in current_group_ids:
+            return self._convert_to_port_model(current)
+
+        updated_group_ids = [
+            sg_id for sg_id in current_group_ids if sg_id != security_group_id
+        ]
+        updated = conn.network.update_port(
+            port_id, security_groups=updated_group_ids
+        )
+        return self._convert_to_port_model(updated)
+
     def create_port(
         self,
         network_id: str,

tests/tools/test_network_tools.py

@@ -869,6 +869,87 @@ def test_set_port_binding_and_admin_state(
         )
         assert res_toggle.is_admin_state_up is True
 
+    def test_add_remove_security_group_on_port(
+        self, mock_openstack_connect_network
+    ):
+        mock_conn = mock_openstack_connect_network
+
+        # current without sg-9
+        current = Mock()
+        current.id = "port-1"
+        current.name = "p1"
+        current.status = "ACTIVE"
+        current.description = None
+        current.project_id = None
+        current.network_id = "net-1"
+        current.admin_state_up = True
+        current.is_admin_state_up = True
+        current.device_id = None
+        current.device_owner = None
+        current.mac_address = "fa:16:3e:00:00:09"
+        current.fixed_ips = []
+        current.security_group_ids = ["sg-1", "sg-2"]
+        mock_conn.network.get_port.return_value = current
+
+        # updated after add
+        updated_add = Mock()
+        updated_add.id = "port-1"
+        updated_add.name = "p1"
+        updated_add.status = "ACTIVE"
+        updated_add.description = None
+        updated_add.project_id = None
+        updated_add.network_id = "net-1"
+        updated_add.admin_state_up = True
+        updated_add.is_admin_state_up = True
+        updated_add.device_id = None
+        updated_add.device_owner = None
+        updated_add.mac_address = "fa:16:3e:00:00:09"
+        updated_add.fixed_ips = []
+        updated_add.security_group_ids = ["sg-1", "sg-2", "sg-9"]
+        mock_conn.network.update_port.return_value = updated_add
+
+        tools = self.get_network_tools()
+        res_add = tools.add_security_group_to_port("port-1", "sg-9")
+        assert isinstance(res_add, Port)
+        mock_conn.network.update_port.assert_called_with(
+            "port-1", security_groups=["sg-1", "sg-2", "sg-9"]
+        )
+
+        # idempotent add when sg already present
+        mock_conn.network.get_port.return_value = updated_add
+        res_add_again = tools.add_security_group_to_port("port-1", "sg-9")
+        assert isinstance(res_add_again, Port)
+
+        # updated after remove
+        updated_remove = Mock()
+        updated_remove.id = "port-1"
+        updated_remove.name = "p1"
+        updated_remove.status = "ACTIVE"
+        updated_remove.description = None
+        updated_remove.project_id = None
+        updated_remove.network_id = "net-1"
+        updated_remove.admin_state_up = True
+        updated_remove.is_admin_state_up = True
+        updated_remove.device_id = None
+        updated_remove.device_owner = None
+        updated_remove.mac_address = "fa:16:3e:00:00:09"
+        updated_remove.fixed_ips = []
+        updated_remove.security_group_ids = ["sg-1", "sg-2"]
+        mock_conn.network.update_port.return_value = updated_remove
+
+        res_remove = tools.remove_security_group_from_port("port-1", "sg-9")
+        assert isinstance(res_remove, Port)
+        mock_conn.network.update_port.assert_called_with(
+            "port-1", security_groups=["sg-1", "sg-2"]
+        )
+
+        # idempotent remove when sg not present
+        mock_conn.network.get_port.return_value = updated_remove
+        res_remove_again = tools.remove_security_group_from_port(
+            "port-1", "sg-9"
+        )
+        assert isinstance(res_remove_again, Port)
+
     def test_get_subnets_filters_and_has_gateway_true(
         self,
         mock_openstack_connect_network,