Been a while but I have been doing some tinkering with Nile’s MAB APIs to manage wired client devices and assign them to certain segments. There are multiple ways to do this:
- RADIUS
- SSO (SCIM)
- Fingerprinting
- MAB
I was working with the MAB APIs to see what kind of automation can be created with it. Steps that my script takes:
- Search for a device
- MAC
- Device Type
- Port
- State
- Next specify which option is being used.
- Script will pull the list of devices based on the option
- Choose the device(s)
- From the list of the Segments, choose the segment device needs to be moved into.
- Confirm and that is it.
Once the device is moved into a segment it can be connected anywhere in a Nile Service Block and it will always end up in that segment. Don’t have to worry about having to reconfigure another switch or a port if someone moves the device. I’d like to bring up few points here.
- I see this as a benefit to the IT teams because they won’t have to worry about chasing devices when they are being moved around and update individual switches and ports.
- Imagine doing a network refresh, don’t have to worry about which cable needs to go where.
- Improved security because now I don’t have to worry about devices ending up in networks they don’t belong (I am thinking possibly a camera or some IoT devices with possible malware ending up in a segment with Finance, Payroll, Health records etc).
- NOTE: Even if someone makes a mistake and puts a device with malware into a sensitive segment, Nile Service Block’s default behavior is to block east/west traffic even within the same network.
Script Used:
This is the script I used, feel free to use/modify it. NOTE: Script is provided as is without any support or warranty, use it at your own risk.
# Want to Log changes for Audit or just becaus
log_filename = "search_device_segment_changes.log"
logging.basicConfig(
filename=log_filename,
level=logging.INFO,
format="%(asctime)s - %(levelname)s - %(message)s"
)
# Load API key
arf_nile_apikey = read_yaml_nilearf()
# Fetch segments and devices (I have two other scripts to do this and this script uses that data)
segments = GET_seg.get_segments()
devices = GET_MAB.get_devices()
# Log the fetch of segments and devices
logging.info("Fetched segments and devices.")
if not segments:
logging.error("No segments found. Check API call in GET_seg.")
print("No segments found. Check API call in GET_seg.")
exit(1)
if not devices:
logging.error("No devices found. Check API call in GET_MAB.")
print("No devices found. Check API call in GET_MAB.")
exit(1)
# Function to filter devices based on search criteria
def search_devices(devices, search_term, search_field="macAddress"):
"""
Filter devices by checking both clientInfo and clientConfig for the field.
"""
filtered = []
for device in devices:
client_info = device.get("clientInfo") or {}
client_config = device.get("clientConfig") or {}
# Look in both sections
value = client_info.get(search_field) or client_config.get(search_field)
if value and search_term.lower() in str(value).lower():
filtered.append(device)
return filtered
# Enter a search term, this will be the actual search term
search_term = input("Enter a search term to filter devices (e.g., MAC, device type, port, state): ").strip()
# This is for the search criteria or defining, look for this term in this field for instance.
search_field = input("Choose the field for search (options: macAddress, deviceType, port, state): ").strip()
# Some error checking and validation
valid_fields = ["macAddress", "deviceType", "port", "state"]
if search_field not in valid_fields:
logging.error(f"Invalid search field: {search_field}")
print(f"Invalid search field: {search_field}. Valid options are: {', '.join(valid_fields)}")
exit(1)
# Filter the devices based on the search term and field
filtered_devices = search_devices(devices, search_term, search_field)
# Display the filtered devices or exit
if not filtered_devices:
logging.error("No devices found matching the search criteria.")
print("No devices found matching the search criteria.")
exit(1)
else:
print("Filtered Devices:")
for i, device in enumerate(filtered_devices):
client_info = device.get('clientInfo') or {}
client_config = device.get('clientConfig') or {}
device_type = client_info.get('deviceType', 'Unknown')
port = client_config.get('port', 'Unknown')
state = client_config.get('state', 'Unknown')
mac = client_info.get('macAddress') or client_config.get('macAddress', 'Unknown')
print(f"{i + 1}: DeviceType: {device_type}, Port: {port}, State: {state}, MAC: {mac}")
# User selects devices
device_indices = input(":arrow_right: Select multiple devices by numbers (comma separated): ")
selected_device_indices = [int(idx.strip()) - 1 for idx in device_indices.split(",")]
# Validate and log selected devices
selected_devices = []
for idx in selected_device_indices:
try:
selected_device = filtered_devices[idx]
selected_devices.append(selected_device)
logging.info(f"User selected device: {selected_device['clientInfo']['macAddress']}")
except IndexError:
logging.error(f"Invalid device index: {idx + 1}")
print(f"Invalid device index: {idx + 1}. Please check your input.")
exit(1)
# Display segments
print("\nAvailable Segments:")
for i, seg in enumerate(segments):
print(f"{i + 1}: {seg['Segment']} (ID: {seg['ID']})")
# User selects segment
segment_index = int(input("Select a segment by number: ")) - 1
selected_segment = segments[segment_index]
# Log segment selection
logging.info(f"User selected segment: {selected_segment['Segment']} (ID: {selected_segment['ID']})")
# Confirmation Summary
print("\n--- Summary ---")
print(f"Selected Devices: {[device['clientInfo']['macAddress'] for device in selected_devices]}")
print(f"New Segment: {selected_segment['Segment']} (ID: {selected_segment['ID']})")
print("State will be updated to: AUTH_OK")
# Log summary
logging.info(f"Selected Devices: {[device['clientInfo']['macAddress'] for device in selected_devices]}")
logging.info(f"Selected Segment: {selected_segment['Segment']} (ID: {selected_segment['ID']})")
logging.info("State will be updated to AUTH_OK.")
confirm = input("Confirm changes for all selected devices? (yes/no): ").strip().lower()
# Log confirmation
if confirm == 'yes':
logging.info("User confirmed changes. Sending PATCH request.")
# Prepare the list of devices to update
macs_list = []
for device in selected_devices:
mac_address = device['clientInfo']['macAddress']
macs_list.append({
"macAddress": mac_address,
"description": None,
"rule": "",
"ruleType": "INDIVIDUAL",
"segmentId": selected_segment['ID'],
"state": "AUTH_OK",
"geoScope": {
"siteIds": [],
"buildingIds": [],
"floorIds": []
},
"staticIp": None,
"silentIp": None,
"ipAddress": ""
})
# Send the PATCH request for bulk update
url = "https://u1.nile-global.cloud/api/v1/client-configs"
payload = {
"macsList": macs_list
}
headers = {
"Content-Type": "application/json",
"x-nile-api-key": arf_nile_apikey
}
response = requests.patch(url, headers=headers, json=payload)
# Log the response
if response.status_code in [200, 204]:
logging.info(f"Bulk update successful. HTTP {response.status_code}")
print(f"Bulk update successful (status: {response.status_code}).")
else:
logging.error(f"Bulk update failed. HTTP {response.status_code} - {response.text}")
print(f"Bulk update failed: {response.status_code} - {response.text}")
else:
logging.warning("User canceled the operation.")
print("Operation cancelled.")
Thanks for reading, would love to hear your feedback and questions.