feat(network): add security group rule mcp tools (#87)

Author: platanus-kr

src/openstack_mcp_server/tools/network_tools.py

@@ -63,6 +63,10 @@ def register_tools(self, mcp: FastMCP):
         mcp.tool()(self.get_security_group_detail)
         mcp.tool()(self.update_security_group)
         mcp.tool()(self.delete_security_group)
+        mcp.tool()(self.create_security_group_rule)
+        mcp.tool()(self.get_security_group_rule_detail)
+        mcp.tool()(self.delete_security_group_rule)
+        mcp.tool()(self.create_security_group_rules_bulk)
 
     def get_networks(
         self,
@@ -1285,11 +1289,16 @@ def _convert_to_security_group_model(self, openstack_sg) -> SecurityGroup:
         rule_ids: list[str] | None = None
         rules = getattr(openstack_sg, "security_group_rules", None)
         if rules is not None:
-            dto_rules = [
-                SecurityGroupRule.model_validate(r, from_attributes=True)
-                for r in rules
-            ]
-            rule_ids = [str(r.id) for r in dto_rules if getattr(r, "id", None)]
+            extracted: list[str] = []
+            for r in rules:
+                rid = (
+                    r.get("id")
+                    if isinstance(r, dict)
+                    else getattr(r, "id", None)
+                )
+                if rid:
+                    extracted.append(str(rid))
+            rule_ids = extracted
 
         return SecurityGroup(
             id=openstack_sg.id,
@@ -1299,3 +1308,116 @@ def _convert_to_security_group_model(self, openstack_sg) -> SecurityGroup:
             project_id=getattr(openstack_sg, "project_id", None),
             security_group_rule_ids=rule_ids,
         )
+
+    def create_security_group_rule(
+        self,
+        security_group_id: str,
+        direction: str = "ingress",
+        ethertype: str = "IPv4",
+        protocol: str | None = None,
+        port_range_min: int | None = None,
+        port_range_max: int | None = None,
+        remote_ip_prefix: str | None = None,
+        remote_group_id: str | None = None,
+        description: str | None = None,
+        project_id: str | None = None,
+    ) -> SecurityGroupRule:
+        """
+        Create a Security Group Rule.
+
+        :param security_group_id: Target security group ID
+        :param direction: "ingress" or "egress"
+        :param ethertype: "IPv4" or "IPv6"
+        :param protocol: L4 protocol (e.g., "tcp", "udp", "icmp")
+        :param port_range_min: Minimum port
+        :param port_range_max: Maximum port
+        :param remote_ip_prefix: Source/destination CIDR
+        :param remote_group_id: Peer security group ID
+        :param description: Rule description
+        :param project_id: Project ownership
+        :return: Created SecurityGroupRule
+        """
+        conn = get_openstack_conn()
+        args: dict = {
+            "security_group_id": security_group_id,
+            "direction": direction,
+            "ethertype": ethertype,
+        }
+        args["protocol"] = protocol
+        args["port_range_min"] = port_range_min
+        args["port_range_max"] = port_range_max
+        args["remote_ip_prefix"] = remote_ip_prefix
+        args["remote_group_id"] = remote_group_id
+        args["description"] = description
+        args["project_id"] = project_id
+        rule = conn.network.create_security_group_rule(**args)
+        return self._convert_to_security_group_rule_model(rule)
+
+    def get_security_group_rule_detail(
+        self, rule_id: str
+    ) -> SecurityGroupRule:
+        """
+        Get detailed information about a specific Security Group Rule.
+
+        :param rule_id: Rule ID
+        :return: SecurityGroupRule detail
+        """
+        conn = get_openstack_conn()
+        rule = conn.network.get_security_group_rule(rule_id)
+        return self._convert_to_security_group_rule_model(rule)
+
+    def delete_security_group_rule(self, rule_id: str) -> None:
+        """
+        Delete a Security Group Rule.
+
+        :param rule_id: Rule ID to delete
+        :return: None
+        """
+        conn = get_openstack_conn()
+        conn.network.delete_security_group_rule(rule_id, ignore_missing=False)
+        return None
+
+    def create_security_group_rules_bulk(
+        self,
+        rules: list[dict],
+    ) -> list[SecurityGroupRule]:
+        """
+        Create multiple Security Group Rules in bulk.
+
+        Each rule dict should follow Neutron SG rule schema keys (e.g.,
+        security_group_id, direction, ethertype, protocol, port_range_min,
+        port_range_max, remote_ip_prefix, remote_group_id, description, project_id).
+
+        :param rules: List of rule dictionaries
+        :return: List of created SecurityGroupRule models
+        """
+        conn = get_openstack_conn()
+        created = conn.network.create_security_group_rules(rules=rules)
+        return [self._convert_to_security_group_rule_model(r) for r in created]
+
+    def _convert_to_security_group_rule_model(
+        self, openstack_rule
+    ) -> SecurityGroupRule:
+        """
+        Convert an OpenStack Security Group Rule object to a pydantic model.
+
+        :param openstack_rule: OpenStack rule object
+        :return: SecurityGroupRule model
+        """
+        return SecurityGroupRule(
+            id=getattr(openstack_rule, "id"),
+            name=getattr(openstack_rule, "name", None),
+            status=getattr(openstack_rule, "status", None),
+            description=getattr(openstack_rule, "description", None),
+            project_id=getattr(openstack_rule, "project_id", None),
+            direction=getattr(openstack_rule, "direction", None),
+            ethertype=getattr(openstack_rule, "ethertype", None),
+            protocol=getattr(openstack_rule, "protocol", None),
+            port_range_min=getattr(openstack_rule, "port_range_min", None),
+            port_range_max=getattr(openstack_rule, "port_range_max", None),
+            remote_ip_prefix=getattr(openstack_rule, "remote_ip_prefix", None),
+            remote_group_id=getattr(openstack_rule, "remote_group_id", None),
+            security_group_id=getattr(
+                openstack_rule, "security_group_id", None
+            ),
+        )

tests/tools/test_network_tools.py

@@ -1768,3 +1768,137 @@ def test_add_get_remove_router_interface_by_port(
         assert removed == RouterInterface(
             router_id="r-if-2", port_id="p-2", subnet_id="s-2"
         )
+
+    def test_create_security_group_rule(self, mock_openstack_connect_network):
+        mock_conn = mock_openstack_connect_network
+        rule = Mock()
+        rule.id = "r-1"
+        rule.name = None
+        rule.status = None
+        rule.description = "allow 22"
+        rule.project_id = "proj-1"
+        rule.direction = "ingress"
+        rule.ethertype = "IPv4"
+        rule.protocol = "tcp"
+        rule.port_range_min = 22
+        rule.port_range_max = 22
+        rule.remote_ip_prefix = "0.0.0.0/0"
+        rule.remote_group_id = None
+        rule.security_group_id = "sg-1"
+        mock_conn.network.create_security_group_rule.return_value = rule
+
+        tools = self.get_network_tools()
+        res = tools.create_security_group_rule(
+            security_group_id="sg-1",
+            direction="ingress",
+            ethertype="IPv4",
+            protocol="tcp",
+            port_range_min=22,
+            port_range_max=22,
+            remote_ip_prefix="0.0.0.0/0",
+            description="allow 22",
+            project_id="proj-1",
+        )
+        assert res.id == "r-1"
+        mock_conn.network.create_security_group_rule.assert_called_once()
+
+    def test_get_security_group_rule_detail(
+        self, mock_openstack_connect_network
+    ):
+        mock_conn = mock_openstack_connect_network
+        rule = Mock()
+        rule.id = "r-2"
+        rule.name = None
+        rule.status = None
+        rule.description = None
+        rule.project_id = None
+        rule.direction = "egress"
+        rule.ethertype = "IPv4"
+        rule.protocol = None
+        rule.port_range_min = None
+        rule.port_range_max = None
+        rule.remote_ip_prefix = None
+        rule.remote_group_id = None
+        rule.security_group_id = "sg-1"
+        mock_conn.network.get_security_group_rule.return_value = rule
+
+        tools = self.get_network_tools()
+        res = tools.get_security_group_rule_detail("r-2")
+        assert res.id == "r-2"
+        mock_conn.network.get_security_group_rule.assert_called_once_with(
+            "r-2"
+        )
+
+    def test_delete_security_group_rule(self, mock_openstack_connect_network):
+        mock_conn = mock_openstack_connect_network
+        mock_conn.network.delete_security_group_rule.return_value = None
+
+        tools = self.get_network_tools()
+        res = tools.delete_security_group_rule("r-3")
+        assert res is None
+        mock_conn.network.delete_security_group_rule.assert_called_once_with(
+            "r-3", ignore_missing=False
+        )
+
+    def test_create_security_group_rules_bulk(
+        self, mock_openstack_connect_network
+    ):
+        mock_conn = mock_openstack_connect_network
+        r1 = Mock()
+        r1.id = "r-10"
+        r1.name = None
+        r1.status = None
+        r1.description = None
+        r1.project_id = None
+        r1.security_group_id = "sg-1"
+        r1.direction = "ingress"
+        r1.ethertype = "IPv4"
+        r1.protocol = "tcp"
+        r1.port_range_min = 80
+        r1.port_range_max = 80
+        r1.remote_ip_prefix = "0.0.0.0/0"
+        r1.remote_group_id = None
+
+        r2 = Mock()
+        r2.id = "r-11"
+        r2.name = None
+        r2.status = None
+        r2.description = None
+        r2.project_id = None
+        r2.security_group_id = "sg-1"
+        r2.direction = "ingress"
+        r2.ethertype = "IPv4"
+        r2.protocol = "tcp"
+        r2.port_range_min = 443
+        r2.port_range_max = 443
+        r2.remote_ip_prefix = "0.0.0.0/0"
+        r2.remote_group_id = None
+
+        mock_conn.network.create_security_group_rules.return_value = [r1, r2]
+
+        tools = self.get_network_tools()
+        rules = [
+            {
+                "security_group_id": "sg-1",
+                "direction": "ingress",
+                "ethertype": "IPv4",
+                "protocol": "tcp",
+                "port_range_min": 80,
+                "port_range_max": 80,
+                "remote_ip_prefix": "0.0.0.0/0",
+            },
+            {
+                "security_group_id": "sg-1",
+                "direction": "ingress",
+                "ethertype": "IPv4",
+                "protocol": "tcp",
+                "port_range_min": 443,
+                "port_range_max": 443,
+                "remote_ip_prefix": "0.0.0.0/0",
+            },
+        ]
+        res = tools.create_security_group_rules_bulk(rules)
+        assert len(res) == 2
+        mock_conn.network.create_security_group_rules.assert_called_once_with(
+            rules=rules
+        )