fix(access_strategy/ip_restriction): only whitelist or blacklist, not… (

#171)

* fix(access_strategy/ip_restriction): only whitelist or blacklist, not both

* fix(ipv6/cidr): support ipv6 cidr like 0:0:0:0:0:ffff:192.1.1.1/96 in ip-restriction

src/dashboard/apigateway/apigateway/common/mixins/models.py

@@ -58,7 +58,7 @@ def config(self):
 
     @config.setter
     def config(self, data):
-        # should be valid json string
+        # should be valid JSON string
         if isinstance(data, str):
             data = json.loads(data)
 

src/dashboard/apigateway/apigateway/controller/crds/release_data/access_strategy.py

@@ -104,24 +104,17 @@ def _to_plugin_config(self, access_strategy: AccessStrategy):
         convert to:
             whitelist: [] / blacklist: []
         """
-        whitelist: List[str] = []
-        blacklist: List[str] = []
-
         config = access_strategy.config
         ip_content_list = [group._ips for group in IPGroup.objects.filter(id__in=config["ip_group_list"])]
         ip_list = self._parse_ip_content_list(ip_content_list)
 
         # the access strategy will be remove soon, so use `allow` and `deny` here directly
         if config["type"] == "allow":
-            whitelist = ip_list
+            return {"whitelist": ip_list}
         elif config["type"] == "deny":
-            blacklist = ip_list
-        # do nothing if type is wrong
+            return {"blacklist": ip_list}
 
-        return {
-            "whitelist": whitelist,
-            "blacklist": blacklist,
-        }
+        raise ValueError("type should be either one of allow or deny for access strategy ip-control")
 
 
 class CorsASC(AccessStrategyConvertor):

src/dashboard/apigateway/apigateway/tests/controller/crds/release_data/test_access_strategy.py

@@ -58,19 +58,19 @@ def test_parse_ip_content_list(self, ip_content_list, expected):
         [
             (
                 {"type": "allow", "ip_group_list": [1]},
-                {"whitelist": ["127.0.0.1"], "blacklist": []},
+                {"whitelist": ["127.0.0.1"]},
             ),
             (
                 {"type": "deny", "ip_group_list": [1]},
-                {"whitelist": [], "blacklist": ["127.0.0.1"]},
+                {"blacklist": ["127.0.0.1"]},
             ),
             (
                 {"type": "allow", "ip_group_list": [2]},
-                {"whitelist": ["127.0.0.1"], "blacklist": []},
+                {"whitelist": ["127.0.0.1"]},
             ),
             (
                 {"type": "allow", "ip_group_list": [1, 2]},
-                {"whitelist": ["127.0.0.1"], "blacklist": []},
+                {"whitelist": ["127.0.0.1"]},
             ),
         ],
     )

src/dashboard/apigateway/apigateway/tests/utils/test_ip.py

@@ -29,6 +29,9 @@ class TestIP:
             ("", []),
             ("1.1.1.1", ["1.1.1.1"]),
             ("1.1.1.1\n2.2.2.2 \r\n#comment\n 1.1.1.1", ["1.1.1.1", "2.2.2.2"]),
+            ("::ffff:192.1.1.1", ["::ffff:192.1.1.1"]),
+            ("::ffff:192.1.1.1/96", ["::ffff:c001:101/96"]),
+            ("0:0:0:0:0:ffff:192.1.1.1/96", ["::ffff:c001:101/96"]),
         ],
     )
     def test_parse_ip_content_to_list(self, ip_content, expected):

src/dashboard/apigateway/apigateway/utils/ip.py

@@ -16,7 +16,7 @@
 # to the current version of the project delivered to anyone in the future.
 #
 
-
+import ipaddress
 from typing import List
 
 
@@ -34,6 +34,17 @@ def parse_ip_content_to_list(ip_content: str) -> List[str]:
         ip_line = ip_line.strip()
         if not ip_line or ip_line.startswith("#"):
             continue
+
+        # http://www.tcpipguide.com/free/t_IPv6IPv4AddressEmbedding-2.htm
+        # >>> ipaddress.ip_interface("::ffff:192.1.1.1/96")
+        #     IPv6Interface('::ffff:c001:101/96')
+        # while the apisix not support the `::ffff:192.1.1.1/96`, we need to convert here
+        ip_line_lower = ip_line.lower()
+        if ip_line_lower.startswith("0:0:0:0:0:ffff:") or ip_line_lower.startswith("::ffff:") and "/" in ip_line_lower:
+            # ipv4 in ipv6
+            ips.add(str(ipaddress.ip_interface(ip_line)))
+            continue
+
         ips.add(ip_line)
 
     return list(ips)