Updated all NPM packages, added ESLint back in and configured linting…

…, added additional automated API tests, fixed React dependencies, added total selected/vnets in Block Networks component, fixed bug when saving External Networks and addressed a few additional JS issues that ESLint revealed

.github/workflows/azure-ipam-testing.yml

@@ -3,6 +3,8 @@ name: Azure IPAM Testing
 run-name: Azure IPAM Deployment & Testing
 
 on:
+  push:
+    branches: [ ipam-pydantic-v2 ]
   pull_request:
     branches: [ main ]
 

.gitignore

@@ -1,5 +1,6 @@
 .env
 .VSCodeCounter
+.eslintcache
 NOTES.md
 TODO.md
 /logs

engine/app/models.py

@@ -365,7 +365,7 @@ class JSONPatch(BaseModel):
 
 VNetsUpdate = Annotated[List[str], None]
 
-ExtNetsUpdate = Annotated[List[ExtNet], None]
+ExtNetsUpdate = Annotated[list[ExtNet], None]
 
 DeleteExtNetReq = Annotated[List[str], None]
 
@@ -541,3 +541,23 @@ class NewVNetCIDR(BaseModel):
     space: str
     block: str
     cidr: str
+
+class CIDRContainer(BaseModel):
+    space: str
+    block: str
+
+class CIDRCheckReq(BaseModel):
+    """DOCSTRING"""
+
+    cidr: IPv4Network
+
+class CIDRCheckRes(BaseModel):
+    """DOCSTRING"""
+
+    name: str
+    id: str
+    resource_group: str
+    subscription_id: UUID
+    tenant_id: UUID
+    prefixes: List[IPv4Network]
+    containers: List[CIDRContainer]

engine/app/routers/argquery.py

@@ -167,6 +167,20 @@
 | project name, id, prefix, vwan_name, vwan_id, resource_group, subscription_id, tenant_id, resv
 """
 
+NET_BASIC = """
+resources
+| where type =~ 'Microsoft.Network/virtualNetworks'
+| project name, id, resourceGroup, subscriptionId, tenantId, prefixes = properties.addressSpace.addressPrefixes
+| union (
+    resources
+    | where type =~ 'microsoft.network/virtualhubs'
+    | where isempty(kind)
+    | project name, id, resourceGroup, subscriptionId, tenantId, prefixes = pack_array(properties.addressPrefix)
+)
+| where subscriptionId !in~ {}
+| project name, id, resource_group = resourceGroup, subscription_id = subscriptionId, tenant_id = tenantId, prefixes
+"""
+
 PRIVATE_ENDPOINT = """
 resources
 | where type =~ 'microsoft.network/networkinterfaces'

engine/app/routers/common/helper.py

@@ -49,13 +49,14 @@ def vnet_fixup(vnet_list):
         vnet['prefixes'] = ipv4_prefixes
         # vnet['prefixes_v6'] = ipv6_prefixes
 
-        for subnet in vnet['subnets']:
-            # Subnet IPv4 & IPv6 prefix
-            ipv4_prefix = subnet['prefix'][0]
-            # ipv6_prefix = subnet['prefix'][1] if len(subnet['prefix']) > 1 else None
-
-            subnet['prefix'] = ipv4_prefix
-            # subnet['prefix_v6'] = ipv6_prefix
+        if 'subnets' in vnet:
+            for subnet in vnet['subnets']:
+                # Subnet IPv4 & IPv6 prefix
+                ipv4_prefix = subnet['prefix'][0]
+                # ipv6_prefix = subnet['prefix'][1] if len(subnet['prefix']) > 1 else None
+
+                subnet['prefix'] = ipv4_prefix
+                # subnet['prefix_v6'] = ipv6_prefix
 
     return vnet_list
 

engine/app/routers/space.py

@@ -57,14 +57,25 @@
     dependencies=[Depends(check_token_expired)]
 )
 
-async def scrub_space_patch(patch):
+async def valid_space_name_update(name, space_name, tenant_id):
+    space_names = await cosmos_query("SELECT VALUE LOWER(c.name) FROM c WHERE c.type = 'space' AND LOWER(c.name) != LOWER('{}')".format(space_name), tenant_id)
+
+    if name.lower() in space_names:
+        raise HTTPException(status_code=400, detail="Updated Space name must be unique.")
+
+    if re.match(SPACE_NAME_REGEX, name):
+        return True
+
+    return False
+
+async def scrub_space_patch(patch, space_name, tenant_id):
     scrubbed_patch = []
 
     allowed_ops = [
         {
             "op": "replace",
             "path": "/name",
-            "valid": SPACE_NAME_REGEX,
+            "valid": valid_space_name_update,
             "error": "Space name can be a maximum of 64 characters and may contain alphanumerics, underscores, hypens, and periods."
         },
         {
@@ -79,18 +90,39 @@ async def scrub_space_patch(patch):
         target = next((x for x in allowed_ops if (x['op'] == item['op'] and x['path'] == item['path'])), None)
 
         if target:
-            if re.match(target['valid'], str(item['value'])):
-                scrubbed_patch.append(item)
+            if isinstance(target['valid'], str):
+                if re.match(target['valid'], str(item['value']), re.IGNORECASE):
+                    scrubbed_patch.append(item)
+                else:
+                    raise HTTPException(status_code=400, detail=target['error'])
+            elif callable(target['valid']):
+                if await target['valid'](item['value'], space_name, tenant_id):
+                    scrubbed_patch.append(item)
+                else:
+                    raise HTTPException(status_code=400, detail=target['error'])
             else:
                 raise HTTPException(status_code=400, detail=target['error'])
 
     return scrubbed_patch
 
-async def valid_block_cidr_update(cidr, space, block_name):
+async def valid_block_name_update(name, space_name, block_name, tenant_id):
+    blocks = await cosmos_query("SELECT VALUE LOWER(t.name) FROM c join t IN c.blocks WHERE c.type = 'space' AND LOWER(c.name) != LOWER('{}')".format(space_name), tenant_id)
+    other_blocks = [x for x in blocks if x != block_name.lower()]
+
+    if name.lower() in other_blocks:
+        raise HTTPException(status_code=400, detail="Updated Block name cannot match existing Blocks within the Space.")
+
+    if re.match(BLOCK_NAME_REGEX, name):
+        return True
+
+    return False
+
+async def valid_block_cidr_update(cidr, space_name, block_name, tenant_id):
     space_cidrs = []
     block_cidrs = []
 
-    target_block = next((x for x in space['blocks'] if x['name'].lower() == block_name.lower()), None)
+    blocks = await cosmos_query("SELECT VALUE t FROM c join t IN c.blocks WHERE c.type = 'space' AND LOWER(c.name) = LOWER('{}')".format(space_name), tenant_id)
+    target_block = next((x for x in blocks if x['name'].lower() == block_name.lower()), None)
 
     if target_block:
         if(cidr == target_block['cidr']):
@@ -99,11 +131,11 @@ async def valid_block_cidr_update(cidr, space, block_name):
         block_network = IPNetwork(cidr)
 
         if(str(block_network.cidr) != cidr):
-            raise HTTPException(status_code=400, detail="Invalid CIDR value, Try '{}' instead.".format(block_network.cidr))
+            raise HTTPException(status_code=400, detail="Invalid CIDR value, try '{}' instead.".format(block_network.cidr))
 
     net_list = await get_network(None, True)
 
-    for block in space['blocks']:
+    for block in blocks:
         if block['name'] != block_name:
             space_cidrs.append(block['cidr'])
         else:
@@ -121,21 +153,21 @@ async def valid_block_cidr_update(cidr, space, block_name):
     block_set = IPSet(block_cidrs)
 
     if space_set & update_set:
-        return False
+        raise HTTPException(status_code=400, detail="Updated CIDR cannot overlap other Block CIDRs within the Space.")
 
     if not block_set.issubset(update_set):
         return False
 
     return True
 
-async def scrub_block_patch(patch, space, block_name):
+async def scrub_block_patch(patch, space_name, block_name, tenant_id):
     scrubbed_patch = []
 
     allowed_ops = [
         {
             "op": "replace",
             "path": "/name",
-            "valid": BLOCK_NAME_REGEX,
+            "valid": valid_block_name_update,
             "error": "Block name can be a maximum of 64 characters and may contain alphanumerics, underscores, hypens, slashes, and periods."
         },
         {
@@ -156,7 +188,7 @@ async def scrub_block_patch(patch, space, block_name):
                 else:
                     raise HTTPException(status_code=400, detail=target['error'])
             elif callable(target['valid']):
-                if await target['valid'](item['value'], space, block_name):
+                if await target['valid'](item['value'], space_name, block_name, tenant_id):
                     scrubbed_patch.append(item)
                 else:
                     raise HTTPException(status_code=400, detail=target['error'])
@@ -296,7 +328,7 @@ async def create_space(
         "id": uuid.uuid4(),
         "type": "space",
         "tenant_id": tenant_id,
-        **space.dict(),
+        **space.model_dump(),
         "blocks": []
     }
 
@@ -438,11 +470,11 @@ async def update_space(
         raise HTTPException(status_code=400, detail="Invalid space name.")
 
     try:
-        patch = jsonpatch.JsonPatch(updates)
+        patch = jsonpatch.JsonPatch(updates.model_dump())
     except jsonpatch.InvalidJsonPatch:
         raise HTTPException(status_code=500, detail="Invalid JSON patch, please review and try again.")
 
-    scrubbed_patch = jsonpatch.JsonPatch(await scrub_space_patch(patch))
+    scrubbed_patch = jsonpatch.JsonPatch(await scrub_space_patch(patch, space, tenant_id))
     update_space = scrubbed_patch.apply(target_space)
 
     await cosmos_replace(target_space, update_space)
@@ -896,11 +928,11 @@ async def update_block(
         raise HTTPException(status_code=400, detail="Invalid block name.")
 
     try:
-        patch = jsonpatch.JsonPatch(updates)
+        patch = jsonpatch.JsonPatch(updates.model_dump())
     except jsonpatch.InvalidJsonPatch:
         raise HTTPException(status_code=500, detail="Invalid JSON patch, please review and try again.")
 
-    scrubbed_patch = jsonpatch.JsonPatch(await scrub_block_patch(patch, target_space, block))
+    scrubbed_patch = jsonpatch.JsonPatch(await scrub_block_patch(patch, space, block, tenant_id))
     scrubbed_patch.apply(update_block, in_place=True)
 
     await cosmos_replace(target_space, update_space)
@@ -1474,7 +1506,7 @@ async def update_block_external_net(
     if not is_admin:
         raise HTTPException(status_code=403, detail="API restricted to admins.")
 
-    external_names = list(map(lambda x: x['name'], externals))
+    external_names = list(map(lambda x: x.name, externals))
     unique_ext_nets = len(set(external_names)) == len(external_names)
 
     if not unique_ext_nets:
@@ -1484,10 +1516,10 @@ async def update_block_external_net(
     invalid_descs = []
 
     for external in externals:
-        if not re.match(EXTERNAL_NAME_REGEX, external['name'], re.IGNORECASE):
+        if not re.match(EXTERNAL_NAME_REGEX, external.name, re.IGNORECASE):
             invalid_names.append(external['name'])
 
-        if not re.match(EXTERNAL_DESC_REGEX, external['desc'], re.IGNORECASE):
+        if not re.match(EXTERNAL_DESC_REGEX, external.desc, re.IGNORECASE):
             invalid_descs.append(external['desc'])
 
     if invalid_names:
@@ -1512,8 +1544,8 @@ async def update_block_external_net(
     external_nets_set = IPSet([])
 
     for external in externals:
-        if not (external_nets_set & IPSet([external['cidr']])):
-            external_nets_set.add(external['cidr'])
+        if not (external_nets_set & IPSet([external.cidr])):
+            external_nets_set.add(external.cidr)
         else:
             external_nets_overlap = True
 

engine/app/routers/tool.py

@@ -5,9 +5,11 @@
     Header
 )
 
+from typing import List
+
 import regex
 import copy
-from netaddr import IPSet
+from netaddr import IPSet, IPNetwork
 
 from app.dependencies import (
     check_token_expired,
@@ -203,4 +205,65 @@ async def next_available_vnet(
 
     return new_cidr
 
+@router.post(
+    "/cidrCheck",
+    summary = "Find Virtual Networks that Overlap a Given CIDR Range",
+    response_model = List[CIDRCheckRes],
+    status_code = 200
+)
+@cosmos_retry(
+    max_retry = 5,
+    error_msg = "Error fetching overlapping virtual networks, please try again."
+)
+async def cidr_check(
+    req: CIDRCheckReq,
+    authorization: str = Header(None, description="Azure Bearer token"),
+    tenant_id: str = Depends(get_tenant_id)
+):
+    """
+    Get a list of Virtual Networks which overlap a given CIDR range with the following information:
+
+    - **cidr**: CIDR range
+
+    <font color='red'>**EXPERIMENTAL**: This API is currently in testing and may change in future releases!</font>
+    """
+
+    if IPNetwork(req.cidr).ip != IPNetwork(req.cidr).network:
+        raise HTTPException(status_code=400, detail="Invalid CIDR range.")
+
+    spaces = await cosmos_query("SELECT * FROM c WHERE c.type = 'space'", tenant_id)
+
+    nets = await arg_query(authorization, True, argquery.NET_BASIC)
+
+    nets = vnet_fixup(nets)
+
+    overlap = [net for net in nets if IPSet([req.cidr]) & IPSet(net['prefixes'])]
+
+    for item in overlap:
+        new_prefixes = []
+
+        for prefix in item['prefixes']:
+            if IPSet([req.cidr]) & IPSet([prefix]):
+                new_prefixes.append(prefix)
+
+        item['prefixes'] = new_prefixes
+
+        item['containers'] = []
+
+        for space in spaces:
+            for block in space['blocks']:
+                for vnet in block['vnets']:
+                    if vnet['id'] == item['id']:
+                        container = {
+                            "space": space['name'],
+                            "block": block['name']
+                        }
+
+                        # container = "/spaces/{}/blocks/{}".format(space['name'], block['name'])
+
+                        item['containers'].append(container)
+
+    return overlap
+
+# Use below for new/experimental APIs:
 # <font color='red'>**EXPERIMENTAL**: This API is currently in testing and may change in future releases!</font>