Remediate the 2026-09-03 feature-test findings

plugins/network/api/routes.py

@@ -3,12 +3,14 @@
 from flask import Blueprint, request
 from flask_jwt_extended import jwt_required
 
+import ipaddress
+
 from shopdb.api import db, Asset, AssetType, Vendor, Communication, CommunicationType, AuditLog, success_response, error_response, paginated_response, ErrorCodes, get_pagination_params, paginate_query
 
 from ..services.assetnumbers import generate_for_type
 from ..models import NetworkDevice, NetworkDeviceType, Subnet, VLAN
 
-from shopdb.api import require_permission, apply_import_timestamps
+from shopdb.api import require_permission, apply_import_timestamps, intarg
 
 network_bp = Blueprint('network', __name__)
 
@@ -232,20 +234,20 @@ def list_network_devices():
         )
 
     # Type filter
-    if type_id := request.args.get('typeid', request.args.get('type_id')):
-        query = query.filter(NetworkDevice.networkdevicetypeid == int(type_id))
+    if (type_id := intarg('typeid', intarg('type_id'))) is not None:
+        query = query.filter(NetworkDevice.networkdevicetypeid == type_id)
 
     # Vendor filter
-    if vendor_id := request.args.get('vendorid', request.args.get('vendor_id')):
-        query = query.filter(NetworkDevice.vendorid == int(vendor_id))
+    if (vendor_id := intarg('vendorid', intarg('vendor_id'))) is not None:
+        query = query.filter(NetworkDevice.vendorid == vendor_id)
 
     # Location filter
-    if location_id := request.args.get('locationid', request.args.get('location_id')):
-        query = query.filter(Asset.locationid == int(location_id))
+    if (location_id := intarg('locationid', intarg('location_id'))) is not None:
+        query = query.filter(Asset.locationid == location_id)
 
     # Business unit filter
-    if bu_id := request.args.get('businessunitid', request.args.get('businessunit_id')):
-        query = query.filter(Asset.businessunitid == int(bu_id))
+    if (bu_id := intarg('businessunitid', intarg('businessunit_id'))) is not None:
+        query = query.filter(Asset.businessunitid == bu_id)
 
     # PoE filter
     if poe := request.args.get('poe'):
@@ -697,11 +699,27 @@ def create_vlan():
     if not data.get('name'):
         return error_response(ErrorCodes.VALIDATION_ERROR, 'name is required')
 
+    # 802.1Q says 1-4094, and nothing checked. 99999 and -5 were stored, and a
+    # non-numeric value was worse: the duplicate probe below compared a string
+    # against an integer column, which MySQL coerces to 0, so 'xyz' matched an
+    # existing VLAN 0 and the error said "VLAN xyz already exists" about a row
+    # that had nothing to do with it.
+    try:
+        vlannumber = int(data['vlannumber'])
+    except (TypeError, ValueError):
+        return error_response(ErrorCodes.VALIDATION_ERROR,
+                              'vlannumber must be a whole number')
+    if not 1 <= vlannumber <= 4094:
+        return error_response(
+            ErrorCodes.VALIDATION_ERROR,
+            'vlannumber must be between 1 and 4094 (802.1Q)')
+    data['vlannumber'] = vlannumber
+
     # Check for duplicate VLAN number
-    if VLAN.query.filter_by(vlannumber=data['vlannumber']).first():
+    if VLAN.query.filter_by(vlannumber=vlannumber).first():
         return error_response(
             ErrorCodes.CONFLICT,
-            f"VLAN {data['vlannumber']} already exists",
+            f"VLAN {vlannumber} already exists",
             http_code=409
         )
 
@@ -784,10 +802,14 @@ def delete_vlan(vlan_id: int):
 
     # Check if VLAN has associated subnets
     if vlan.subnets.filter_by(isactive=True).count() > 0:
+        # 409 like every other in-use guard in the product (network types,
+        # machine types, vendors). It answered 400, which tells the caller
+        # their request was malformed when the request was fine and the state
+        # was the problem.
         return error_response(
-            ErrorCodes.VALIDATION_ERROR,
+            ErrorCodes.CONFLICT,
             'Cannot delete VLAN with associated subnets',
-            http_code=400
+            http_code=409
         )
 
     vlan.isactive = False
@@ -827,12 +849,12 @@ def list_subnets():
         )
 
     # VLAN filter
-    if vlan_id := request.args.get('vlanid'):
-        query = query.filter(Subnet.vlanid == int(vlan_id))
+    if (vlan_id := intarg('vlanid')) is not None:
+        query = query.filter(Subnet.vlanid == vlan_id)
 
     # Location filter
-    if location_id := request.args.get('locationid'):
-        query = query.filter(Subnet.locationid == int(location_id))
+    if (location_id := intarg('locationid')) is not None:
+        query = query.filter(Subnet.locationid == location_id)
 
     # Type filter
     if subnet_type := request.args.get('type'):
@@ -936,10 +958,30 @@ def create_subnet():
     if not data.get('name'):
         return error_response(ErrorCodes.VALIDATION_ERROR, 'name is required')
 
-    # Validate CIDR format (basic check)
-    cidr = data['cidr']
+    # Parsed, not eyeballed for a slash. 'notanip/xx' passed the old check and
+    # was stored; _devices_in_subnet then caught the ValueError from
+    # ip_network and returned an empty device list, so the subnet looked real
+    # and permanently empty. ipaddress is the same library that has to read it
+    # later, so agreeing with it is the whole test.
+    cidr = (data['cidr'] or '').strip()
+    # An explicit prefix is required. ipaddress reads a bare '10.1.1.0' as a
+    # /32, which is legal and almost never what somebody typing a SUBNET meant
+    # - it would silently create a subnet holding one address.
     if '/' not in cidr:
-        return error_response(ErrorCodes.VALIDATION_ERROR, 'cidr must be in CIDR notation (e.g., 10.1.1.0/24)')
+        return error_response(
+            ErrorCodes.VALIDATION_ERROR,
+            'cidr needs a prefix length, e.g. 10.1.1.0/24')
+    try:
+        network = ipaddress.ip_network(cidr, strict=False)
+    except ValueError as bad:
+        return error_response(
+            ErrorCodes.VALIDATION_ERROR,
+            'cidr must be a network in CIDR notation, e.g. 10.1.1.0/24 (%s)'
+            % bad)
+    # Store the normalised form so 10.1.1.5/24 and 10.1.1.0/24 are one subnet
+    # rather than two rows nobody can tell apart.
+    cidr = str(network)
+    data['cidr'] = cidr
 
     # Check for duplicate CIDR
     if Subnet.query.filter_by(cidr=cidr).first():