afce24379c
A sweep of the URL map found 165 GET routes that answered anybody, with no token at all: the whole asset inventory, every machine, PC, printer and network device, the USB checkout history and who is holding which drive, the staff directory with its photographs, every core report. Guest access had been argued about one feature-flag at a time while the front door stood open. Each of those now carries the read permission it belongs to. That also makes the guest role mean something: a site can open exactly the reads it chooses, and nothing follows by accident. Thirty-three endpoints stay public, and each is named with its reason rather than left to be inferred: the app shell and its bootstrap, the settings allowlist (the handler already filters an anonymous caller down to branding and a few keys), images fetched by <img src> which carries no Authorization header, what a display kiosk renders before anybody signs in, the parts and USB kiosks, the client scripts that run as SYSTEM before a user session exists, and a secure-send link whose unguessable token IS the authorisation. tests/test_core/test_public_read_surface.py is the standing gate, the companion to the mutating-route sweep that has guarded writes for months. A new public read fails CI until somebody adds it to that list in front of a reviewer. Four further reads need a login but no permission, because the handler decides from who you are - your own user record, your own API tokens, the role catalogue the grid renders - and the same test proves each still refuses an anonymous caller. Verified not vacuous by ungating one route and watching the sweep name it. The employees plugin gained employees.view; the directory is names, SSOs, photographs and reporting lines, and it had no permission of any kind. Around ninety tests asserted payload shapes against endpoints that answered anybody. They are about what the answer says, not about who may ask, so they are signed in now rather than the gate being loosened. The ones that assert an anonymous caller is refused were left exactly as they were.
1062 lines
37 KiB
Python
1062 lines
37 KiB
Python
"""Network plugin API endpoints."""
|
|
|
|
from flask import Blueprint, request
|
|
from flask_jwt_extended import jwt_required
|
|
|
|
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
|
|
|
|
network_bp = Blueprint('network', __name__)
|
|
|
|
|
|
def _primary_ip(assetid):
|
|
"""The asset's primary IP address string, or None. Mirrors the printer
|
|
route: a device's IP lives in a Communication row, not on the extension."""
|
|
comm = (Communication.query.filter_by(assetid=assetid, isprimary=True).first()
|
|
or Communication.query.filter_by(assetid=assetid).first())
|
|
return comm.ipaddress if comm else None
|
|
|
|
|
|
def _upsert_primary_ip(asset, ip):
|
|
"""Create/update/clear the asset's primary IP Communication from an
|
|
ipaddress string. No-op if the IP CommunicationType is not seeded."""
|
|
ip = (ip or '').strip()
|
|
comm = Communication.query.filter_by(assetid=asset.assetid, isprimary=True).first()
|
|
if not ip:
|
|
# Clearing the field clears the row's IP - it used to return early, so
|
|
# an address could be typed but never taken back out.
|
|
if comm:
|
|
comm.ipaddress = None
|
|
return
|
|
if comm:
|
|
comm.ipaddress = ip
|
|
return
|
|
ip_comtype = CommunicationType.query.filter_by(comtype='IP').first()
|
|
if ip_comtype:
|
|
db.session.add(Communication(
|
|
assetid=asset.assetid, comtypeid=ip_comtype.comtypeid,
|
|
ipaddress=ip, isprimary=True))
|
|
|
|
|
|
# =============================================================================
|
|
# Network Device Types
|
|
# =============================================================================
|
|
|
|
@network_bp.route('/types', methods=['GET'])
|
|
@require_permission('network.view')
|
|
def list_network_device_types():
|
|
"""List all network device types."""
|
|
page, per_page = get_pagination_params(request)
|
|
|
|
query = NetworkDeviceType.query
|
|
|
|
if request.args.get('active', 'true').lower() != 'false':
|
|
query = query.filter(NetworkDeviceType.isactive == True)
|
|
|
|
if search := request.args.get('search'):
|
|
query = query.filter(NetworkDeviceType.networkdevicetype.ilike(f'%{search}%'))
|
|
|
|
query = query.order_by(NetworkDeviceType.networkdevicetype)
|
|
|
|
items, total = paginate_query(query, page, per_page)
|
|
data = [t.to_dict() for t in items]
|
|
|
|
return paginated_response(data, page, per_page, total)
|
|
|
|
|
|
@network_bp.route('/types/<int:type_id>', methods=['GET'])
|
|
@require_permission('network.view')
|
|
def get_network_device_type(type_id: int):
|
|
"""Get a single network device type."""
|
|
t = db.session.get(NetworkDeviceType, type_id)
|
|
|
|
if not t:
|
|
return error_response(
|
|
ErrorCodes.NOT_FOUND,
|
|
f'Network device type with ID {type_id} not found',
|
|
http_code=404
|
|
)
|
|
|
|
return success_response(t.to_dict())
|
|
|
|
|
|
@network_bp.route('/types', methods=['POST'])
|
|
@require_permission('network.create')
|
|
def create_network_device_type():
|
|
"""Create a new network device type."""
|
|
data = request.get_json()
|
|
|
|
if not data or not data.get('networkdevicetype'):
|
|
return error_response(ErrorCodes.VALIDATION_ERROR, 'networkdevicetype is required')
|
|
|
|
existing = NetworkDeviceType.query.filter_by(networkdevicetype=data['networkdevicetype']).first()
|
|
if existing:
|
|
if not existing.isactive:
|
|
existing.isactive = True
|
|
for key in ('description', 'icon', 'color'):
|
|
if data.get(key) is not None:
|
|
setattr(existing, key, data[key])
|
|
db.session.commit()
|
|
return success_response(existing.to_dict(), message='Reactivated existing type')
|
|
return error_response(
|
|
ErrorCodes.CONFLICT,
|
|
f"Network device type '{data['networkdevicetype']}' already exists",
|
|
http_code=409
|
|
)
|
|
|
|
t = NetworkDeviceType(
|
|
networkdevicetype=data['networkdevicetype'],
|
|
description=data.get('description'),
|
|
icon=data.get('icon'), color=data.get('color'),
|
|
prefix=data.get('prefix')
|
|
)
|
|
|
|
db.session.add(t)
|
|
db.session.commit()
|
|
|
|
return success_response(t.to_dict(), message='Network device type created', http_code=201)
|
|
|
|
|
|
@network_bp.route('/types/<int:type_id>', methods=['PUT'])
|
|
@require_permission('network.edit')
|
|
def update_network_device_type(type_id: int):
|
|
"""Update a network device type."""
|
|
t = db.session.get(NetworkDeviceType, type_id)
|
|
|
|
if not t:
|
|
return error_response(
|
|
ErrorCodes.NOT_FOUND,
|
|
f'Network device type with ID {type_id} not found',
|
|
http_code=404
|
|
)
|
|
|
|
data = request.get_json()
|
|
if not data:
|
|
return error_response(ErrorCodes.VALIDATION_ERROR, 'No data provided')
|
|
|
|
if 'networkdevicetype' in data and data['networkdevicetype'] != t.networkdevicetype:
|
|
if NetworkDeviceType.query.filter_by(networkdevicetype=data['networkdevicetype']).first():
|
|
return error_response(
|
|
ErrorCodes.CONFLICT,
|
|
f"Network device type '{data['networkdevicetype']}' already exists",
|
|
http_code=409
|
|
)
|
|
|
|
for key in ['networkdevicetype', 'description', 'icon', 'color', 'prefix',
|
|
'isactive']:
|
|
if key in data:
|
|
setattr(t, key, data[key])
|
|
|
|
db.session.commit()
|
|
return success_response(t.to_dict(), message='Network device type updated')
|
|
|
|
|
|
@network_bp.route('/types/<int:type_id>', methods=['DELETE'])
|
|
@require_permission('network.delete')
|
|
def delete_network_device_type(type_id: int):
|
|
"""Delete a network device type. Refused if any device still uses it."""
|
|
t = db.session.get(NetworkDeviceType, type_id)
|
|
if not t:
|
|
return error_response(ErrorCodes.NOT_FOUND, 'Network device type not found', http_code=404)
|
|
inuse = NetworkDevice.query.filter_by(networkdevicetypeid=type_id).count()
|
|
if inuse:
|
|
return error_response(ErrorCodes.CONFLICT,
|
|
f"Cannot delete: {inuse} device(s) still use this type", http_code=409)
|
|
db.session.delete(t)
|
|
db.session.commit()
|
|
return success_response(message='Network device type deleted')
|
|
|
|
|
|
# =============================================================================
|
|
# Network Devices CRUD
|
|
# =============================================================================
|
|
|
|
@network_bp.route('', methods=['GET'])
|
|
@require_permission('network.view')
|
|
def list_network_devices():
|
|
"""
|
|
List all network devices with filtering and pagination.
|
|
|
|
Query parameters:
|
|
- page, per_page: Pagination
|
|
- active: Filter by active status
|
|
- search: Search by asset number, name, or hostname
|
|
- type_id: Filter by network device type ID
|
|
- vendor_id: Filter by vendor ID
|
|
- location_id: Filter by location ID
|
|
- businessunit_id: Filter by business unit ID
|
|
- poe: Filter by PoE capability (true/false)
|
|
- managed: Filter by managed status (true/false)
|
|
"""
|
|
page, per_page = get_pagination_params(request)
|
|
|
|
# Join NetworkDevice with Asset
|
|
query = db.session.query(NetworkDevice).join(Asset)
|
|
|
|
# Active filter
|
|
if request.args.get('active', 'true').lower() != 'false':
|
|
query = query.filter(Asset.isactive == True)
|
|
|
|
# Exact-match natural-key lookup for idempotent import (asset number).
|
|
if exactassetnumber := request.args.get('assetnumber'):
|
|
query = query.filter(Asset.assetnumber == exactassetnumber)
|
|
|
|
# Search filter. Type and vendor are columns in the list, so both must be
|
|
# searchable. Outer joins so a device missing either still matches on its
|
|
# own fields.
|
|
if search := request.args.get('search'):
|
|
pattern = f'%{search}%'
|
|
query = query.outerjoin(
|
|
NetworkDeviceType,
|
|
NetworkDevice.networkdevicetypeid == NetworkDeviceType.networkdevicetypeid
|
|
).outerjoin(
|
|
Vendor, NetworkDevice.vendorid == Vendor.vendorid
|
|
).filter(
|
|
db.or_(
|
|
Asset.assetnumber.ilike(pattern),
|
|
Asset.name.ilike(pattern),
|
|
Asset.serialnumber.ilike(pattern),
|
|
# The optional identifiers too (ADR-001). Global search matches
|
|
# these, and a tag read off the machine has to find it here as
|
|
# well - this box is where someone holding the label looks.
|
|
Asset.gaugelabreference.ilike(pattern),
|
|
Asset.maintenancereference.ilike(pattern),
|
|
NetworkDevice.hostname.ilike(pattern),
|
|
NetworkDeviceType.networkdevicetype.ilike(pattern),
|
|
Vendor.vendor.ilike(pattern)
|
|
)
|
|
)
|
|
|
|
# Type filter
|
|
if type_id := request.args.get('typeid', request.args.get('type_id')):
|
|
query = query.filter(NetworkDevice.networkdevicetypeid == int(type_id))
|
|
|
|
# Vendor filter
|
|
if vendor_id := request.args.get('vendorid', request.args.get('vendor_id')):
|
|
query = query.filter(NetworkDevice.vendorid == int(vendor_id))
|
|
|
|
# Location filter
|
|
if location_id := request.args.get('locationid', request.args.get('location_id')):
|
|
query = query.filter(Asset.locationid == int(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))
|
|
|
|
# PoE filter
|
|
if poe := request.args.get('poe'):
|
|
query = query.filter(NetworkDevice.ispoe == (poe.lower() == 'true'))
|
|
|
|
# Managed filter
|
|
if managed := request.args.get('managed'):
|
|
query = query.filter(NetworkDevice.ismanaged == (managed.lower() == 'true'))
|
|
|
|
# Sorting
|
|
sort_by = request.args.get('sort', 'hostname')
|
|
sort_dir = request.args.get('dir', 'asc')
|
|
|
|
if sort_by == 'hostname':
|
|
col = NetworkDevice.hostname
|
|
elif sort_by == 'assetnumber':
|
|
col = Asset.assetnumber
|
|
elif sort_by == 'name':
|
|
col = Asset.name
|
|
else:
|
|
col = NetworkDevice.hostname
|
|
|
|
query = query.order_by(col.desc() if sort_dir == 'desc' else col)
|
|
|
|
items, total = paginate_query(query, page, per_page)
|
|
|
|
# Build response with both asset and network device data
|
|
data = []
|
|
for netdev in items:
|
|
item = netdev.asset.to_dict() if netdev.asset else {}
|
|
item['networkdevice'] = netdev.to_dict()
|
|
item['ipaddress'] = _primary_ip(netdev.assetid)
|
|
data.append(item)
|
|
|
|
return paginated_response(data, page, per_page, total)
|
|
|
|
|
|
@network_bp.route('/<int:device_id>', methods=['GET'])
|
|
@require_permission('network.view')
|
|
def get_network_device(device_id: int):
|
|
"""Get a single network device with full details."""
|
|
netdev = db.session.get(NetworkDevice, device_id)
|
|
|
|
if not netdev:
|
|
return error_response(
|
|
ErrorCodes.NOT_FOUND,
|
|
f'Network device with ID {device_id} not found',
|
|
http_code=404
|
|
)
|
|
|
|
result = netdev.asset.to_dict() if netdev.asset else {}
|
|
result['networkdevice'] = netdev.to_dict()
|
|
result['ipaddress'] = _primary_ip(netdev.assetid)
|
|
|
|
return success_response(result)
|
|
|
|
|
|
@network_bp.route('/by-asset/<int:asset_id>', methods=['GET'])
|
|
@require_permission('network.view')
|
|
def get_network_device_by_asset(asset_id: int):
|
|
"""Get network device data by asset ID."""
|
|
netdev = NetworkDevice.query.filter_by(assetid=asset_id).first()
|
|
|
|
if not netdev:
|
|
return error_response(
|
|
ErrorCodes.NOT_FOUND,
|
|
f'Network device for asset {asset_id} not found',
|
|
http_code=404
|
|
)
|
|
|
|
result = netdev.asset.to_dict() if netdev.asset else {}
|
|
result['networkdevice'] = netdev.to_dict()
|
|
result['ipaddress'] = _primary_ip(netdev.assetid)
|
|
|
|
return success_response(result)
|
|
|
|
|
|
@network_bp.route('/by-hostname/<hostname>', methods=['GET'])
|
|
@require_permission('network.view')
|
|
def get_network_device_by_hostname(hostname: str):
|
|
"""Get network device by hostname."""
|
|
netdev = NetworkDevice.query.filter_by(hostname=hostname).first()
|
|
|
|
if not netdev:
|
|
return error_response(
|
|
ErrorCodes.NOT_FOUND,
|
|
f'Network device with hostname {hostname} not found',
|
|
http_code=404
|
|
)
|
|
|
|
result = netdev.asset.to_dict() if netdev.asset else {}
|
|
result['networkdevice'] = netdev.to_dict()
|
|
result['ipaddress'] = _primary_ip(netdev.assetid)
|
|
|
|
return success_response(result)
|
|
|
|
|
|
@network_bp.route('', methods=['POST'])
|
|
@require_permission('network.create')
|
|
def create_network_device():
|
|
"""
|
|
Create new network device (creates both Asset and NetworkDevice records).
|
|
|
|
Required fields:
|
|
- assetnumber: Business identifier
|
|
|
|
Optional fields:
|
|
- name, serialnumber, statusid, locationid, businessunitid
|
|
- networkdevicetypeid, vendorid, modelnumberid, hostname
|
|
- firmwareversion, portcount, ispoe, ismanaged, rackunit
|
|
- mapx, mapy, notes
|
|
"""
|
|
data = request.get_json()
|
|
|
|
if not data:
|
|
return error_response(ErrorCodes.VALIDATION_ERROR, 'No data provided')
|
|
|
|
# Generate from the type prefix + name when left blank. Only when blank: a
|
|
# device carrying a real identifier of its own - vendor tag, controller
|
|
# name, serial - keeps it, which is the platform rule.
|
|
if not data.get('assetnumber'):
|
|
data['assetnumber'] = generate_for_type(
|
|
data.get('networkdevicetypeid'), data.get('name'))
|
|
if not data.get('assetnumber'):
|
|
return error_response(
|
|
ErrorCodes.VALIDATION_ERROR,
|
|
'assetnumber is required (or give a name, and a device type with a '
|
|
'prefix, to have one generated)')
|
|
|
|
# Check for duplicate assetnumber
|
|
if Asset.query.filter_by(assetnumber=data['assetnumber']).first():
|
|
return error_response(
|
|
ErrorCodes.CONFLICT,
|
|
f"Asset with number '{data['assetnumber']}' already exists",
|
|
http_code=409
|
|
)
|
|
|
|
# Check for duplicate hostname
|
|
if data.get('hostname'):
|
|
if NetworkDevice.query.filter_by(hostname=data['hostname']).first():
|
|
return error_response(
|
|
ErrorCodes.CONFLICT,
|
|
f"Network device with hostname '{data['hostname']}' already exists",
|
|
http_code=409
|
|
)
|
|
|
|
# Get network device asset type
|
|
network_type = AssetType.query.filter_by(assettype='network_device').first()
|
|
if not network_type:
|
|
return error_response(
|
|
ErrorCodes.INTERNAL_ERROR,
|
|
'Network device asset type not found. Plugin may not be properly installed.',
|
|
http_code=500
|
|
)
|
|
|
|
# Create the core asset
|
|
asset = Asset(
|
|
assetnumber=data['assetnumber'],
|
|
name=data.get('name'),
|
|
serialnumber=data.get('serialnumber'),
|
|
gaugelabreference=data.get('gaugelabreference'),
|
|
maintenancereference=data.get('maintenancereference'),
|
|
assettypeid=network_type.assettypeid,
|
|
statusid=data.get('statusid', 1),
|
|
locationid=data.get('locationid'),
|
|
businessunitid=data.get('businessunitid'),
|
|
mapx=data.get('mapx'),
|
|
levelid=data.get('levelid'),
|
|
mapy=data.get('mapy'),
|
|
notes=data.get('notes')
|
|
)
|
|
|
|
db.session.add(asset)
|
|
db.session.flush() # Get the assetid
|
|
|
|
# Create the network device extension
|
|
netdev = NetworkDevice(
|
|
assetid=asset.assetid,
|
|
networkdevicetypeid=data.get('networkdevicetypeid'),
|
|
vendorid=data.get('vendorid'),
|
|
modelnumberid=data.get('modelnumberid'),
|
|
hostname=data.get('hostname'),
|
|
firmwareversion=data.get('firmwareversion'),
|
|
portcount=data.get('portcount'),
|
|
ispoe=data.get('ispoe', False),
|
|
ismanaged=data.get('ismanaged', False),
|
|
rackunit=data.get('rackunit')
|
|
)
|
|
|
|
db.session.add(netdev)
|
|
db.session.flush()
|
|
|
|
# Attach the primary IP as a Communication when supplied (a device's IP is
|
|
# not a NetworkDevice column). Mirrors the printer create route.
|
|
if data.get('ipaddress'):
|
|
_upsert_primary_ip(asset, data['ipaddress'])
|
|
|
|
# Preserve legacy timestamps in import mode (no-op otherwise)
|
|
apply_import_timestamps(asset, data)
|
|
|
|
# Audit log
|
|
AuditLog.log('created', 'NetworkDevice', entityid=netdev.networkdeviceid,
|
|
entityname=data.get('hostname') or data['assetnumber'])
|
|
|
|
db.session.commit()
|
|
|
|
result = asset.to_dict()
|
|
result['networkdevice'] = netdev.to_dict()
|
|
result['ipaddress'] = _primary_ip(asset.assetid)
|
|
|
|
return success_response(result, message='Network device created', http_code=201)
|
|
|
|
|
|
@network_bp.route('/<int:device_id>', methods=['PUT'])
|
|
@require_permission('network.edit')
|
|
def update_network_device(device_id: int):
|
|
"""Update network device (both Asset and NetworkDevice records)."""
|
|
netdev = db.session.get(NetworkDevice, device_id)
|
|
|
|
if not netdev:
|
|
return error_response(
|
|
ErrorCodes.NOT_FOUND,
|
|
f'Network device with ID {device_id} not found',
|
|
http_code=404
|
|
)
|
|
|
|
data = request.get_json()
|
|
if not data:
|
|
return error_response(ErrorCodes.VALIDATION_ERROR, 'No data provided')
|
|
|
|
asset = netdev.asset
|
|
|
|
# An UPDATE may not blank the asset number. Create validates it and the
|
|
# column is NOT NULL, but the conflict check below only runs when the value
|
|
# DIFFERS, and '' never collides with anything - so a payload carrying an
|
|
# empty assetnumber wrote it straight through. A form that loaded blank
|
|
# (a failed reference load, a partial fetch) then saved the blank over a
|
|
# real record.
|
|
if 'assetnumber' in data and not (data['assetnumber'] or '').strip():
|
|
return error_response(
|
|
ErrorCodes.VALIDATION_ERROR,
|
|
'assetnumber cannot be empty')
|
|
|
|
# Check for conflicting assetnumber
|
|
if 'assetnumber' in data and data['assetnumber'] != asset.assetnumber:
|
|
if Asset.query.filter_by(assetnumber=data['assetnumber']).first():
|
|
return error_response(
|
|
ErrorCodes.CONFLICT,
|
|
f"Asset with number '{data['assetnumber']}' already exists",
|
|
http_code=409
|
|
)
|
|
|
|
# Check for conflicting hostname
|
|
if 'hostname' in data and data['hostname'] != netdev.hostname:
|
|
existing = NetworkDevice.query.filter_by(hostname=data['hostname']).first()
|
|
if existing and existing.networkdeviceid != device_id:
|
|
return error_response(
|
|
ErrorCodes.CONFLICT,
|
|
f"Network device with hostname '{data['hostname']}' already exists",
|
|
http_code=409
|
|
)
|
|
|
|
# Track changes for audit log
|
|
changes = {}
|
|
|
|
# Update asset fields
|
|
asset_fields = ['assetnumber', 'name', 'serialnumber', 'gaugelabreference',
|
|
'maintenancereference', 'statusid',
|
|
'locationid', 'businessunitid', 'mapx', 'mapy', 'levelid', 'notes', 'isactive']
|
|
for key in asset_fields:
|
|
if key in data:
|
|
old_val = getattr(asset, key)
|
|
new_val = data[key]
|
|
if old_val != new_val:
|
|
changes[key] = {'old': old_val, 'new': new_val}
|
|
setattr(asset, key, data[key])
|
|
|
|
# Update network device fields
|
|
netdev_fields = ['networkdevicetypeid', 'vendorid', 'modelnumberid', 'hostname',
|
|
'firmwareversion', 'portcount', 'ispoe', 'ismanaged', 'rackunit']
|
|
for key in netdev_fields:
|
|
if key in data:
|
|
old_val = getattr(netdev, key)
|
|
new_val = data[key]
|
|
if old_val != new_val:
|
|
changes[key] = {'old': old_val, 'new': new_val}
|
|
setattr(netdev, key, data[key])
|
|
|
|
# Audit log if there were changes
|
|
if changes:
|
|
AuditLog.log('updated', 'NetworkDevice', entityid=netdev.networkdeviceid,
|
|
entityname=netdev.hostname or asset.assetnumber, changes=changes)
|
|
|
|
# Upsert the primary IP Communication when an ipaddress is supplied.
|
|
if 'ipaddress' in data:
|
|
_upsert_primary_ip(asset, data['ipaddress'])
|
|
|
|
apply_import_timestamps(asset, data)
|
|
db.session.commit()
|
|
|
|
result = asset.to_dict()
|
|
result['networkdevice'] = netdev.to_dict()
|
|
result['ipaddress'] = _primary_ip(asset.assetid)
|
|
|
|
return success_response(result, message='Network device updated')
|
|
|
|
|
|
@network_bp.route('/<int:device_id>', methods=['DELETE'])
|
|
@require_permission('network.delete')
|
|
def delete_network_device(device_id: int):
|
|
"""Delete (soft delete) network device."""
|
|
netdev = db.session.get(NetworkDevice, device_id)
|
|
|
|
if not netdev:
|
|
return error_response(
|
|
ErrorCodes.NOT_FOUND,
|
|
f'Network device with ID {device_id} not found',
|
|
http_code=404
|
|
)
|
|
|
|
# Soft delete the asset
|
|
netdev.asset.isactive = False
|
|
|
|
# Audit log
|
|
AuditLog.log('deleted', 'NetworkDevice', entityid=netdev.networkdeviceid,
|
|
entityname=netdev.hostname or netdev.asset.assetnumber)
|
|
|
|
db.session.commit()
|
|
|
|
return success_response(message='Network device deleted')
|
|
|
|
|
|
# =============================================================================
|
|
# Dashboard
|
|
# =============================================================================
|
|
|
|
@network_bp.route('/dashboard/summary', methods=['GET'])
|
|
@require_permission('network.view')
|
|
def dashboard_summary():
|
|
"""Get network device dashboard summary data."""
|
|
# Total active network devices
|
|
total = db.session.query(NetworkDevice).join(Asset).filter(
|
|
Asset.isactive == True
|
|
).count()
|
|
|
|
# Count by device type
|
|
by_type = db.session.query(
|
|
NetworkDeviceType.networkdevicetype,
|
|
db.func.count(NetworkDevice.networkdeviceid)
|
|
).join(NetworkDevice, NetworkDevice.networkdevicetypeid == NetworkDeviceType.networkdevicetypeid
|
|
).join(Asset, Asset.assetid == NetworkDevice.assetid
|
|
).filter(Asset.isactive == True
|
|
).group_by(NetworkDeviceType.networkdevicetype
|
|
).all()
|
|
|
|
# Count by vendor
|
|
by_vendor = db.session.query(
|
|
Vendor.vendor,
|
|
db.func.count(NetworkDevice.networkdeviceid)
|
|
).join(NetworkDevice, NetworkDevice.vendorid == Vendor.vendorid
|
|
).join(Asset, Asset.assetid == NetworkDevice.assetid
|
|
).filter(Asset.isactive == True
|
|
).group_by(Vendor.vendor
|
|
).all()
|
|
|
|
# Count PoE vs non-PoE
|
|
poe_count = db.session.query(NetworkDevice).join(Asset).filter(
|
|
Asset.isactive == True,
|
|
NetworkDevice.ispoe == True
|
|
).count()
|
|
|
|
return success_response({
|
|
'total': total,
|
|
'bytype': [{'type': t, 'count': c} for t, c in by_type],
|
|
'byvendor': [{'vendor': v, 'count': c} for v, c in by_vendor],
|
|
'poe': poe_count,
|
|
'nonpoe': total - poe_count
|
|
})
|
|
|
|
|
|
# =============================================================================
|
|
# VLANs
|
|
# =============================================================================
|
|
|
|
@network_bp.route('/vlans', methods=['GET'])
|
|
@require_permission('network.view')
|
|
def list_vlans():
|
|
"""List all VLANs with filtering and pagination."""
|
|
page, per_page = get_pagination_params(request)
|
|
|
|
query = VLAN.query
|
|
|
|
if request.args.get('active', 'true').lower() != 'false':
|
|
query = query.filter(VLAN.isactive == True)
|
|
|
|
# Search filter
|
|
if search := request.args.get('search'):
|
|
query = query.filter(
|
|
db.or_(
|
|
VLAN.name.ilike(f'%{search}%'),
|
|
VLAN.description.ilike(f'%{search}%'),
|
|
db.cast(VLAN.vlannumber, db.String).ilike(f'%{search}%')
|
|
)
|
|
)
|
|
|
|
# Type filter
|
|
if vlan_type := request.args.get('type'):
|
|
query = query.filter(VLAN.vlantype == vlan_type)
|
|
|
|
query = query.order_by(VLAN.vlannumber)
|
|
|
|
items, total = paginate_query(query, page, per_page)
|
|
data = [v.to_dict() for v in items]
|
|
|
|
return paginated_response(data, page, per_page, total)
|
|
|
|
|
|
@network_bp.route('/vlans/<int:vlan_id>', methods=['GET'])
|
|
@require_permission('network.view')
|
|
def get_vlan(vlan_id: int):
|
|
"""Get a single VLAN with its subnets."""
|
|
vlan = db.session.get(VLAN, vlan_id)
|
|
|
|
if not vlan:
|
|
return error_response(
|
|
ErrorCodes.NOT_FOUND,
|
|
f'VLAN with ID {vlan_id} not found',
|
|
http_code=404
|
|
)
|
|
|
|
result = vlan.to_dict()
|
|
# Include associated subnets
|
|
result['subnets'] = [s.to_dict() for s in vlan.subnets.filter_by(isactive=True).all()]
|
|
|
|
return success_response(result)
|
|
|
|
|
|
@network_bp.route('/vlans', methods=['POST'])
|
|
@require_permission('network.create')
|
|
def create_vlan():
|
|
"""Create a new VLAN."""
|
|
data = request.get_json()
|
|
|
|
if not data:
|
|
return error_response(ErrorCodes.VALIDATION_ERROR, 'No data provided')
|
|
|
|
if not data.get('vlannumber'):
|
|
return error_response(ErrorCodes.VALIDATION_ERROR, 'vlannumber is required')
|
|
if not data.get('name'):
|
|
return error_response(ErrorCodes.VALIDATION_ERROR, 'name is required')
|
|
|
|
# Check for duplicate VLAN number
|
|
if VLAN.query.filter_by(vlannumber=data['vlannumber']).first():
|
|
return error_response(
|
|
ErrorCodes.CONFLICT,
|
|
f"VLAN {data['vlannumber']} already exists",
|
|
http_code=409
|
|
)
|
|
|
|
vlan = VLAN(
|
|
vlannumber=data['vlannumber'],
|
|
name=data['name'],
|
|
description=data.get('description'),
|
|
vlantype=data.get('vlantype')
|
|
)
|
|
|
|
db.session.add(vlan)
|
|
db.session.flush()
|
|
|
|
# Audit log
|
|
AuditLog.log('created', 'VLAN', entityid=vlan.vlanid,
|
|
entityname=f"VLAN {vlan.vlannumber} - {vlan.name}")
|
|
|
|
db.session.commit()
|
|
|
|
return success_response(vlan.to_dict(), message='VLAN created', http_code=201)
|
|
|
|
|
|
@network_bp.route('/vlans/<int:vlan_id>', methods=['PUT'])
|
|
@require_permission('network.edit')
|
|
def update_vlan(vlan_id: int):
|
|
"""Update a VLAN."""
|
|
vlan = db.session.get(VLAN, vlan_id)
|
|
|
|
if not vlan:
|
|
return error_response(
|
|
ErrorCodes.NOT_FOUND,
|
|
f'VLAN with ID {vlan_id} not found',
|
|
http_code=404
|
|
)
|
|
|
|
data = request.get_json()
|
|
if not data:
|
|
return error_response(ErrorCodes.VALIDATION_ERROR, 'No data provided')
|
|
|
|
# Check for conflicting VLAN number
|
|
if 'vlannumber' in data and data['vlannumber'] != vlan.vlannumber:
|
|
if VLAN.query.filter_by(vlannumber=data['vlannumber']).first():
|
|
return error_response(
|
|
ErrorCodes.CONFLICT,
|
|
f"VLAN {data['vlannumber']} already exists",
|
|
http_code=409
|
|
)
|
|
|
|
# Track changes for audit log
|
|
changes = {}
|
|
for key in ['vlannumber', 'name', 'description', 'vlantype', 'isactive']:
|
|
if key in data:
|
|
old_val = getattr(vlan, key)
|
|
new_val = data[key]
|
|
if old_val != new_val:
|
|
changes[key] = {'old': old_val, 'new': new_val}
|
|
setattr(vlan, key, data[key])
|
|
|
|
# Audit log if there were changes
|
|
if changes:
|
|
AuditLog.log('updated', 'VLAN', entityid=vlan.vlanid,
|
|
entityname=f"VLAN {vlan.vlannumber} - {vlan.name}", changes=changes)
|
|
|
|
db.session.commit()
|
|
return success_response(vlan.to_dict(), message='VLAN updated')
|
|
|
|
|
|
@network_bp.route('/vlans/<int:vlan_id>', methods=['DELETE'])
|
|
@require_permission('network.delete')
|
|
def delete_vlan(vlan_id: int):
|
|
"""Delete (soft delete) a VLAN."""
|
|
vlan = db.session.get(VLAN, vlan_id)
|
|
|
|
if not vlan:
|
|
return error_response(
|
|
ErrorCodes.NOT_FOUND,
|
|
f'VLAN with ID {vlan_id} not found',
|
|
http_code=404
|
|
)
|
|
|
|
# Check if VLAN has associated subnets
|
|
if vlan.subnets.filter_by(isactive=True).count() > 0:
|
|
return error_response(
|
|
ErrorCodes.VALIDATION_ERROR,
|
|
'Cannot delete VLAN with associated subnets',
|
|
http_code=400
|
|
)
|
|
|
|
vlan.isactive = False
|
|
|
|
# Audit log
|
|
AuditLog.log('deleted', 'VLAN', entityid=vlan.vlanid,
|
|
entityname=f"VLAN {vlan.vlannumber} - {vlan.name}")
|
|
|
|
db.session.commit()
|
|
|
|
return success_response(message='VLAN deleted')
|
|
|
|
|
|
# =============================================================================
|
|
# Subnets
|
|
# =============================================================================
|
|
|
|
@network_bp.route('/subnets', methods=['GET'])
|
|
@require_permission('network.view')
|
|
def list_subnets():
|
|
"""List all subnets with filtering and pagination."""
|
|
page, per_page = get_pagination_params(request)
|
|
|
|
query = Subnet.query
|
|
|
|
if request.args.get('active', 'true').lower() != 'false':
|
|
query = query.filter(Subnet.isactive == True)
|
|
|
|
# Search filter
|
|
if search := request.args.get('search'):
|
|
query = query.filter(
|
|
db.or_(
|
|
Subnet.cidr.ilike(f'%{search}%'),
|
|
Subnet.name.ilike(f'%{search}%'),
|
|
Subnet.description.ilike(f'%{search}%')
|
|
)
|
|
)
|
|
|
|
# VLAN filter
|
|
if vlan_id := request.args.get('vlanid'):
|
|
query = query.filter(Subnet.vlanid == int(vlan_id))
|
|
|
|
# Location filter
|
|
if location_id := request.args.get('locationid'):
|
|
query = query.filter(Subnet.locationid == int(location_id))
|
|
|
|
# Type filter
|
|
if subnet_type := request.args.get('type'):
|
|
query = query.filter(Subnet.subnettype == subnet_type)
|
|
|
|
query = query.order_by(Subnet.cidr)
|
|
|
|
items, total = paginate_query(query, page, per_page)
|
|
data = [s.to_dict() for s in items]
|
|
|
|
return paginated_response(data, page, per_page, total)
|
|
|
|
|
|
@network_bp.route('/subnets/<int:subnet_id>', methods=['GET'])
|
|
@require_permission('network.view')
|
|
def get_subnet(subnet_id: int):
|
|
"""Get a single subnet plus the network devices whose primary IP falls in
|
|
its CIDR range."""
|
|
subnet = db.session.get(Subnet, subnet_id)
|
|
|
|
if not subnet:
|
|
return error_response(
|
|
ErrorCodes.NOT_FOUND,
|
|
f'Subnet with ID {subnet_id} not found',
|
|
http_code=404
|
|
)
|
|
|
|
data = subnet.to_dict()
|
|
data['devices'] = _devices_in_subnet(subnet.cidr)
|
|
return success_response(data)
|
|
|
|
|
|
def _devices_in_subnet(cidr):
|
|
"""ANY asset (PC, printer, network device, ...) whose primary IP falls in
|
|
cidr. A subnet is cross-type, so this matches on the core Communication +
|
|
Asset tables, not just network devices. Matching is in Python because the IP
|
|
lives in a Communication row, not a column."""
|
|
import ipaddress
|
|
from shopdb.api import Asset, AssetType, Communication
|
|
try:
|
|
net = ipaddress.ip_network(cidr, strict=False)
|
|
except (ValueError, TypeError):
|
|
return []
|
|
rows = (Communication.query
|
|
.join(Asset, Communication.assetid == Asset.assetid)
|
|
.join(AssetType, Asset.assettypeid == AssetType.assettypeid)
|
|
.filter(Communication.isprimary == True,
|
|
Communication.ipaddress.isnot(None))
|
|
.with_entities(Asset.assetid, Asset.assetnumber, Asset.name,
|
|
AssetType.assettype, Communication.ipaddress).all())
|
|
result = []
|
|
for assetid, assetnumber, name, assettype, ip in rows:
|
|
ip = (ip or '').strip()
|
|
if not ip:
|
|
continue
|
|
try:
|
|
if ipaddress.ip_address(ip) in net:
|
|
result.append({
|
|
'assetid': assetid, 'assetnumber': assetnumber, 'name': name,
|
|
'assettype': assettype, 'ipaddress': ip,
|
|
'url': _asset_detail_url(assettype, assetid),
|
|
})
|
|
except (ValueError, TypeError):
|
|
continue
|
|
result.sort(key=lambda d: d['ipaddress'])
|
|
return result
|
|
|
|
|
|
def _asset_detail_url(assettype, assetid):
|
|
"""Best-effort front-end detail path for an asset by type. Resolves the
|
|
per-type extension id lazily/guarded (subnet listings span plugins); returns
|
|
None when the plugin is absent so the frontend just shows the row."""
|
|
try:
|
|
if assettype == 'network_device':
|
|
row = NetworkDevice.query.filter_by(assetid=assetid).first()
|
|
return f'/network/{row.networkdeviceid}' if row else None
|
|
if assettype == 'printer':
|
|
from plugins.printers.models import Printer
|
|
row = Printer.query.filter_by(assetid=assetid).first()
|
|
return f'/printers/{row.printerid}' if row else None
|
|
if assettype == 'computer':
|
|
from plugins.computers.models import Computer
|
|
row = Computer.query.filter_by(assetid=assetid).first()
|
|
return f'/pcs/{row.computerid}' if row else None
|
|
except ImportError:
|
|
return None
|
|
return None
|
|
|
|
|
|
@network_bp.route('/subnets', methods=['POST'])
|
|
@require_permission('network.create')
|
|
def create_subnet():
|
|
"""Create a new subnet."""
|
|
data = request.get_json()
|
|
|
|
if not data:
|
|
return error_response(ErrorCodes.VALIDATION_ERROR, 'No data provided')
|
|
|
|
if not data.get('cidr'):
|
|
return error_response(ErrorCodes.VALIDATION_ERROR, 'cidr is required')
|
|
if not data.get('name'):
|
|
return error_response(ErrorCodes.VALIDATION_ERROR, 'name is required')
|
|
|
|
# Validate CIDR format (basic check)
|
|
cidr = data['cidr']
|
|
if '/' not in cidr:
|
|
return error_response(ErrorCodes.VALIDATION_ERROR, 'cidr must be in CIDR notation (e.g., 10.1.1.0/24)')
|
|
|
|
# Check for duplicate CIDR
|
|
if Subnet.query.filter_by(cidr=cidr).first():
|
|
return error_response(
|
|
ErrorCodes.CONFLICT,
|
|
f"Subnet {cidr} already exists",
|
|
http_code=409
|
|
)
|
|
|
|
# Validate VLAN if provided
|
|
if data.get('vlanid'):
|
|
if not db.session.get(VLAN, data['vlanid']):
|
|
return error_response(
|
|
ErrorCodes.VALIDATION_ERROR,
|
|
f"VLAN with ID {data['vlanid']} not found"
|
|
)
|
|
|
|
subnet = Subnet(
|
|
cidr=cidr,
|
|
name=data['name'],
|
|
description=data.get('description'),
|
|
gatewayip=data.get('gatewayip'),
|
|
subnetmask=data.get('subnetmask'),
|
|
networkaddress=data.get('networkaddress'),
|
|
broadcastaddress=data.get('broadcastaddress'),
|
|
vlanid=data.get('vlanid'),
|
|
subnettype=data.get('subnettype'),
|
|
locationid=data.get('locationid'),
|
|
dhcpenabled=data.get('dhcpenabled', True),
|
|
dhcprangestart=data.get('dhcprangestart'),
|
|
dhcprangeend=data.get('dhcprangeend'),
|
|
dns1=data.get('dns1'),
|
|
dns2=data.get('dns2')
|
|
)
|
|
|
|
db.session.add(subnet)
|
|
db.session.flush()
|
|
|
|
# Audit log
|
|
AuditLog.log('created', 'Subnet', entityid=subnet.subnetid,
|
|
entityname=f"{subnet.cidr} - {subnet.name}")
|
|
|
|
db.session.commit()
|
|
|
|
return success_response(subnet.to_dict(), message='Subnet created', http_code=201)
|
|
|
|
|
|
@network_bp.route('/subnets/<int:subnet_id>', methods=['PUT'])
|
|
@require_permission('network.edit')
|
|
def update_subnet(subnet_id: int):
|
|
"""Update a subnet."""
|
|
subnet = db.session.get(Subnet, subnet_id)
|
|
|
|
if not subnet:
|
|
return error_response(
|
|
ErrorCodes.NOT_FOUND,
|
|
f'Subnet with ID {subnet_id} not found',
|
|
http_code=404
|
|
)
|
|
|
|
data = request.get_json()
|
|
if not data:
|
|
return error_response(ErrorCodes.VALIDATION_ERROR, 'No data provided')
|
|
|
|
# Check for conflicting CIDR
|
|
if 'cidr' in data and data['cidr'] != subnet.cidr:
|
|
if Subnet.query.filter_by(cidr=data['cidr']).first():
|
|
return error_response(
|
|
ErrorCodes.CONFLICT,
|
|
f"Subnet {data['cidr']} already exists",
|
|
http_code=409
|
|
)
|
|
|
|
allowed_fields = ['cidr', 'name', 'description', 'gatewayip', 'subnetmask',
|
|
'networkaddress', 'broadcastaddress', 'vlanid', 'subnettype',
|
|
'locationid', 'dhcpenabled', 'dhcprangestart', 'dhcprangeend',
|
|
'dns1', 'dns2', 'isactive']
|
|
|
|
# Track changes for audit log
|
|
changes = {}
|
|
for key in allowed_fields:
|
|
if key in data:
|
|
old_val = getattr(subnet, key)
|
|
new_val = data[key]
|
|
if old_val != new_val:
|
|
changes[key] = {'old': old_val, 'new': new_val}
|
|
setattr(subnet, key, data[key])
|
|
|
|
# Audit log if there were changes
|
|
if changes:
|
|
AuditLog.log('updated', 'Subnet', entityid=subnet.subnetid,
|
|
entityname=f"{subnet.cidr} - {subnet.name}", changes=changes)
|
|
|
|
db.session.commit()
|
|
return success_response(subnet.to_dict(), message='Subnet updated')
|
|
|
|
|
|
@network_bp.route('/subnets/<int:subnet_id>', methods=['DELETE'])
|
|
@require_permission('network.delete')
|
|
def delete_subnet(subnet_id: int):
|
|
"""Delete (soft delete) a subnet."""
|
|
subnet = db.session.get(Subnet, subnet_id)
|
|
|
|
if not subnet:
|
|
return error_response(
|
|
ErrorCodes.NOT_FOUND,
|
|
f'Subnet with ID {subnet_id} not found',
|
|
http_code=404
|
|
)
|
|
|
|
subnet.isactive = False
|
|
|
|
# Audit log
|
|
AuditLog.log('deleted', 'Subnet', entityid=subnet.subnetid,
|
|
entityname=f"{subnet.cidr} - {subnet.name}")
|
|
|
|
db.session.commit()
|
|
|
|
return success_response(message='Subnet deleted')
|