vh/NetAlertX
1
0
Fork 0
mirror of https://github.com/jokob-sk/NetAlertX.git synced 2025-12-07 01:26:11 -08:00
Code Issues Actions 3 Packages Projects Releases Wiki Activity

Merge pull request #737 from FlyingToto/main

Browse Source
This commit is contained in:
jokob-sk
2024-07-18 17:25:41 +10:00
committed by GitHub
parent a19e268ea7 74b2432729
commit 4a3598e840
9 changed files with 1830 additions and 34 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
.env.omada.ffsb42 Symbolic link
View File

@@ -0,0 +1 @@
../.env.omada.ffsb42

1
docker-compose.yml.ffsb42 Symbolic link
View File

@@ -0,0 +1 @@
../docker-compose.yml.ffsb42

60
front/plugins/omada_sdn_imp/README.md
View File

@@ -1,18 +1,33 @@
## Overview
Plugin functionality overview and links to external resources if relevant. Include use cases if available.
The OMADA SDN plugin aims at synchronizing data between NetAlertX and a TPLINK OMADA SND controler by leveraging a tplink omada python library.
#### features:
1. extract list of OMADA Clients from OMADA and sync them up with NetAlertX
2. extract list of OAMDA Devices (switches and access points) and sync them up with NetAlertX
> [!TIP]
> Some tip.
> some omada devices are apparently not fully compatible with the API which might lead to partial results.
### Quick setup guide
To set up the plugin correctly, make sure...
1. You SHOULD (ie: strongly recommend) set up an account in your OMADA SDN console dedicated to NetAlertX OMADA_SDN plugin.
- you should set USER TYPE = Local USer
- you should set USER ROLE = Administrator (if you use a read-only role you won't be able to sync names from NetAlerX to OMADA SDN)
- you can set Site Privileges = All Sites (or limit it to specific sites )
2. populate the variables in NetAlertX as instructed in the config plugin page.
#### Required Settings
- When to run `PREF_RUN`
-
- OMDSDN_url
- OMDSDN_sites
- OMDSDN_username
- OMDSDN_password
- OMDSDN_force_overwrite
### Usage
@@ -20,6 +35,37 @@ To set up the plugin correctly, make sure...
### Notes
- Additional notes, limitations, Author info.
#### features not implemented yet:
3. extract list of OAMDA router Devices (er605...) and sync them up with NetAlertX
(I need to setup my own er605 however due to its limitations I have no use for it, and due to limitations of opensense dhcp servers, I can't deploy it yet without breaking dhcp self registration into opnsense unbound - see below)
#### know limitations:
OMADA SDN limitation fixed by the plugin:
0. OMADA SDN can't use DNS for names and keep using MAC ref: https://community.tp-link.com/en/business/forum/topic/503782
- when you use an OMADA user Role = Administrator, the plugin will attempt to fix OMADA's shortcoming and populat the NAME field from NetAlertX (from DNS/DHCP/...)
![OMADA SDN account page](Oamadaomada_sdn_imp.png)
-
Made with ❤ by [@FlyingToto](https://github.com/FlyingToto) 🙏
can not fix some of tplinks OMADA SDN own limitations/bugs:
1. OMADA SDN switches uplinks/downlinks is broken if the default router is not an OMADA native device
- (I try to circumvent that through a tree parsing heuristic but your mileage might vary...)
- ref: https://community.tp-link.com/en/business/forum/topic/673628
2. OMADA SDN clients are sometimes mapped to the wrong switch/port... for instance:
- client -> access_switch1/port1 -> core_switch2/port2 sometimes shows as client -> core_switch2/port2
- it is unclear if this issue is realted to (1)
3. OMADA er605 routers do not self register DHCP names with a remote unbound DNS (nor embded DNS):
- ref: https://community.tp-link.com/en/business/forum/topic/542472
- it looks like some release candidate firmware might provide this feature... I will test when opnsesne Kea self registration get fixed as well(4)and I can get it dhcp proxy to work.
4. Opnsense dhcp doesn't support relay and self-registration at the same time...
- opnsense legacy ISC dhcp server doesn't support dhcp proxies: ref: https://forum.opnsense.org/index.php?topic=34254.0
- opnsense new kea dhcp server doesn't support dns self registration (yet) ref: https://github.com/opnsense/core/pull/7362
5. incompatible devices:
- OMADA EAP245 - to be fair to tp-link, this access point works inside OMADA SDN, so it might be an issue with our omada python library but we can't extract data from it.
### Other infos
- Author : Flying Toto
- Date : 04-Jul-2024 - version 1.0

8
front/plugins/omada_sdn_imp/config.json
View File

@@ -2,7 +2,7 @@
"code_name": "omada_sdn_imp",
"unique_prefix": "OMDSDN",
"plugin_type": "device_scanner",
"execution_order" : "Layer_1",
"execution_order" : "Layer_0",
"enabled": true,
"data_source": "script",
"mapped_to_table": "CurrentScan",
@@ -125,7 +125,7 @@
"description": [
{
"language_code": "en_us",
"string": "Enter full URL with protocol <code>https://CHANGEME_omada.mylocaldomain</code>."
"string": "Enter full URL with protocol <code>https://CHANGEME_omada.mylocaldomain:PORT</code>."
}
]
},
@@ -200,7 +200,7 @@
"description": [
{
"language_code": "en_us",
"string": "Omada SDN site IDs. You can get it by..."
"string": "Omada SDN site IDs. For now, we only process the first site listed since NetAlertX's other probes won't traverse across NAT and routers. But if needed please submit an issue in github with your specific use case for consideration: <code>https://github.com/jokob-sk/NetAlertX/issues </code> "
}
]
},
@@ -282,7 +282,7 @@
"description": [
{
"language_code": "en_us",
"string": "The plugin synchronizes names from NetAlertX to OMADA. By default NetAlertX will only populate missing names in OMADASDN devices (i.e.: where the name is defaulting to the device MAC address); with this setting toggled, it will overwrite existing values regardless."
"string": "The plugin synchronizes names from NetAlertX to OMADA Clietnts. By default NetAlertX will only populate missing names in OMADASDN devices (i.e.: where the name is defaulting to the device MAC address); with this setting toggled, it will overwrite existing values regardless."
}
]
},

690
front/plugins/omada_sdn_imp/config.json.v6 Executable file
View File

@@ -0,0 +1,690 @@
{
"code_name": "omada_sdn_imp",
"unique_prefix": "OMDSDN",
"plugin_type": "device_scanner",
"enabled": true,
"data_source": "script",
"mapped_to_table": "CurrentScan",
"data_filters": [
{
"compare_column": "Object_PrimaryID",
"compare_operator": "==",
"compare_field_id": "txtMacFilter",
"compare_js_template": "'{value}'.toString()",
"compare_use_quotes": true
}
],
"show_ui": true,
"localized": ["display_name", "description", "icon"],
"display_name": [
{
"language_code": "en_us",
"string": "OMADA SDN import"
}
],
"description": [
{
"language_code": "en_us",
"string": "Plugin to import data from OMADA SDN."
}
],
"icon": [
{
"language_code": "en_us",
"string": "<i class=\"fa fa-search\"></i>"
}
],
"params": [],
"settings": [
{
"function": "RUN",
"events": ["run"],
"type": {
"dataType": "string",
"elements": [
{ "elementType": "select", "elementOptions": [], "transformers": [] }
]
},
"default_value": "disabled",
"options": ["disabled", "once", "schedule", "always_after_scan"],
"localized": ["name", "description"],
"name": [
{
"language_code": "en_us",
"string": "When to run"
},
{
"language_code": "es_es",
"string": "Cuándo ejecutar"
},
{
"language_code": "de_de",
"string": "Wann laufen"
}
],
"description": [
{
"language_code": "en_us",
"string": "When the scan should run. Good options are: <code>schedule</code>"
}
]
},
{
"function": "RUN_SCHD",
"type": {
"dataType": "string",
"elements": [
{ "elementType": "input", "elementOptions": [], "transformers": [] }
]
},
"default_value": "*/5 * * * *",
"options": [],
"localized": ["name", "description"],
"name": [
{
"language_code": "en_us",
"string": "Schedule"
}
],
"description": [
{
"language_code": "en_us",
"string": "Only enabled if you select <code>schedule</code> in the <a href=\"#SYNC_RUN\"><code>SYNC_RUN</code> setting</a>. Make sure you enter the schedule in the correct cron-like format (e.g. validate at <a href=\"https://crontab.guru/\" target=\"_blank\">crontab.guru</a>). For example entering <code>0 4 * * *</code> will run the scan after 4 am in the <a onclick=\"toggleAllSettings()\" href=\"#TIMEZONE\"><code>TIMEZONE</code> you set above</a>. Will be run NEXT time the time passes."
},
{
"language_code": "es_es",
"string": "Solo está habilitado si selecciona <code>schedule</code> en la configuración <a href=\"#SYNC_RUN\"><code>SYNC_RUN</code></a>. Asegúrese de ingresar la programación en el formato similar a cron correcto (por ejemplo, valide en <a href=\"https://crontab.guru/\" target=\"_blank\">crontab.guru</a>). Por ejemplo, ingresar <code>0 4 * * *</code> ejecutará el escaneo después de las 4 a.m. en el <a onclick=\"toggleAllSettings()\" href=\"#TIMEZONE\"><code>TIMEZONE</ código> que configuró arriba</a>. Se ejecutará la PRÓXIMA vez que pase el tiempo."
},
{
"language_code": "de_de",
"string": "Nur aktiviert, wenn Sie <code>schedule</code> in der <a href=\"#SYNC_RUN\"><code>SYNC_RUN</code>-Einstellung</a> auswählen. Stellen Sie sicher, dass Sie den Zeitplan im richtigen Cron-ähnlichen Format eingeben (z. B. validieren unter <a href=\"https://crontab.guru/\" target=\"_blank\">crontab.guru</a>). Wenn Sie beispielsweise <code>0 4 * * *</code> eingeben, wird der Scan nach 4 Uhr morgens in der <a onclick=\"toggleAllSettings()\" href=\"#TIMEZONE\"><code>TIMEZONE</ ausgeführt. Code> den Sie oben festgelegt haben</a>. Wird das NÄCHSTE Mal ausgeführt, wenn die Zeit vergeht."
}
]
},
{
"function": "url",
"type": {
"dataType": "string",
"elements": [
{ "elementType": "input", "elementOptions": [], "transformers": [] }
]
},
"maxLength": 50,
"default_value": "",
"options": [],
"localized": ["name", "description"],
"name": [
{
"language_code": "en_us",
"string": "URL"
}
],
"description": [
{
"language_code": "en_us",
"string": "Enter full URL with protocol <code>https://CHANGEME_omada.mylocaldomain</code>."
}
]
},
{
"function": "sites",
"type": {
"dataType": "array",
"elements": [
{
"elementType": "input",
"elementOptions": [
{ "placeholder": "Enter value" },
{ "suffix": "_in" },
{ "cssClasses": "col-sm-10" },
{ "prefillValue": "null" }
],
"transformers": []
},
{
"elementType": "button",
"elementOptions": [
{ "sourceSuffixes": ["_in"] },
{ "separator": "" },
{ "cssClasses": "col-xs-12" },
{ "onClick": "addList(this, false)" },
{ "getStringKey": "Gen_Add" }
],
"transformers": []
},
{
"elementType": "button",
"elementOptions": [
{ "sourceSuffixes": [] },
{ "separator": "" },
{ "cssClasses": "col-xs-6" },
{ "onClick": "removeAllOptions(this)" },
{ "getStringKey": "Gen_Remove_All" }
],
"transformers": []
},
{
"elementType": "button",
"elementOptions": [
{ "sourceSuffixes": [] },
{ "separator": "" },
{ "cssClasses": "col-xs-6" },
{ "onClick": "removeFromList(this)" },
{ "getStringKey": "Gen_Remove_Last" }
],
"transformers": []
},
{
"elementType": "select",
"elementOptions": [
{ "multiple": "true" },
{ "readonly": "true" },
{ "editable": "true" }
],
"transformers": []
}
]
},
"default_value": [],
"options": [],
"localized": ["name", "description"],
"name": [
{
"language_code": "en_us",
"string": "OMADA sites"
}
],
"description": [
{
"language_code": "en_us",
"string": "Omada SDN site IDs. You can get it by..."
}
]
},
{
"function": "username",
"type": {
"dataType": "string",
"elements": [
{ "elementType": "input", "elementOptions": [], "transformers": [] }
]
},
"maxLength": 50,
"default_value": "",
"options": [],
"localized": ["name", "description"],
"name": [
{
"language_code": "en_us",
"string": "User name"
}
],
"description": [
{
"language_code": "en_us",
"string": "Omada SDN instance user name."
}
]
},
{
"function": "password",
"type": {
"dataType": "string",
"elements": [
{
"elementType": "input",
"elementOptions": [{ "type": "password" }],
"transformers": []
}
]
},
"maxLength": 50,
"default_value": "",
"options": [],
"localized": ["name", "description"],
"name": [
{
"language_code": "en_us",
"string": "Password"
}
],
"description": [
{
"language_code": "en_us",
"string": "Omada SDN instance password"
}
]
},
{
"function": "force_overwrite",
"type": {
"dataType": "boolean",
"elements": [
{
"elementType": "input",
"elementOptions": [{ "type": "checkbox" }],
"transformers": []
}
]
},
"default_value": false,
"options": [],
"localized": ["name", "description"],
"name": [
{
"language_code": "en_us",
"string": "Force overwrite"
}
],
"description": [
{
"language_code": "en_us",
"string": "The plugin synchronizes names from NetAlertX to OMADA. By default NetAlertX will only populate missing names in OMADASDN devices (i.e.: where the name is defaulting to the device MAC address); with this setting toggled, it will overwrite existing values regardless."
}
]
},
{
"function": "CMD",
"type": {
"dataType": "string",
"elements": [
{
"elementType": "input",
"elementOptions": [{ "readonly": "true" }],
"transformers": []
}
]
},
"default_value": "python3 /app/front/plugins/omada_sdn_imp/omada_sdn.py",
"options": [],
"localized": ["name", "description"],
"name": [
{
"language_code": "en_us",
"string": "Command"
},
{
"language_code": "es_es",
"string": "Comando"
},
{
"language_code": "de_de",
"string": "Befehl"
}
],
"description": [
{
"language_code": "en_us",
"string": "Command to run. This can not be changed"
},
{
"language_code": "es_es",
"string": "Comando a ejecutar. Esto no se puede cambiar"
},
{
"language_code": "de_de",
"string": "Befehl zum Ausführen. Dies kann nicht geändert werden"
}
]
},
{
"function": "RUN_TIMEOUT",
"type": {
"dataType": "integer",
"elements": [
{
"elementType": "input",
"elementOptions": [{ "type": "number" }],
"transformers": []
}
]
},
"default_value": 30,
"options": [],
"localized": ["name", "description"],
"name": [
{
"language_code": "en_us",
"string": "Run timeout"
},
{
"language_code": "es_es",
"string": "Tiempo límite de ejecución"
},
{
"language_code": "de_de",
"string": "Zeitüberschreitung"
}
],
"description": [
{
"language_code": "en_us",
"string": "Maximum time in seconds to wait for the script to finish. If this time is exceeded the script is aborted."
},
{
"language_code": "es_es",
"string": "Tiempo máximo en segundos para esperar a que finalice el script. Si se supera este tiempo, el script se cancela."
},
{
"language_code": "de_de",
"string": "Maximale Zeit in Sekunden, die auf den Abschluss des Skripts gewartet werden soll. Bei Überschreitung dieser Zeit wird das Skript abgebrochen."
}
]
},
{
"default_value": [],
"description": [
{
"language_code": "en_us",
"string": "Send a notification if selected values change. Use <code>CTRL + Click</code> to select/deselect. <ul> <li><code>Watched_Value1</code> is Hostname </li><li><code>Watched_Value2</code> is Parent Node </li><li><code>Watched_Value3</code> is Port </li><li><code>Watched_Value4</code> is SSID </li></ul>"
}
],
"function": "WATCH",
"localized": ["name", "description"],
"name": [
{
"language_code": "en_us",
"string": "Watched"
},
{
"language_code": "es_es",
"string": "Visto"
}
],
"options": [
"Watched_Value1",
"Watched_Value2",
"Watched_Value3",
"Watched_Value4"
],
"type": {
"dataType": "array",
"elements": [
{
"elementType": "select",
"elementOptions": [{ "multiple": "true" }],
"transformers": []
}
]
}
},
{
"default_value": ["new", "watched-changed"],
"description": [
{
"language_code": "en_us",
"string": "Send a notification only on these statuses. <code>new</code> means a new unique (unique combination of PrimaryId and SecondaryId) object was discovered. <code>watched-changed</code> means that selected <code>Watched_ValueN</code> columns changed."
}
],
"function": "REPORT_ON",
"localized": ["name", "description"],
"name": [
{
"language_code": "en_us",
"string": "Report on"
}
],
"options": [
"new",
"watched-changed",
"watched-not-changed",
"missing-in-last-scan"
],
"type": {
"dataType": "array",
"elements": [
{
"elementType": "select",
"elementOptions": [{ "multiple": "true" }],
"transformers": []
}
]
}
}
],
"database_column_definitions": [
{
"column": "Object_PrimaryID",
"mapped_to_column": "cur_MAC",
"css_classes": "col-sm-2",
"show": true,
"type": "device_name_mac",
"default_value": "",
"options": [],
"localized": ["name"],
"name": [
{
"language_code": "en_us",
"string": "MAC"
},
{
"language_code": "es_es",
"string": "MAC"
},
{
"language_code": "de_de",
"string": "MAC"
}
]
},
{
"column": "Object_SecondaryID",
"mapped_to_column": "cur_IP",
"css_classes": "col-sm-2",
"show": true,
"type": "device_ip",
"default_value": "",
"options": [],
"localized": ["name"],
"name": [
{
"language_code": "en_us",
"string": "IP"
},
{
"language_code": "es_es",
"string": "IP"
},
{
"language_code": "de_de",
"string": "IP"
}
]
},
{
"column": "Watched_Value1",
"mapped_to_column": "cur_Name",
"css_classes": "col-sm-2",
"show": true,
"type": "label",
"default_value": "",
"options": [],
"localized": ["name"],
"name": [
{
"language_code": "en_us",
"string": "Name"
}
]
},
{
"column": "Watched_Value2",
"mapped_to_column": "cur_NetworkNodeMAC",
"css_classes": "col-sm-2",
"show": true,
"type": "label",
"default_value": "",
"options": [],
"localized": ["name"],
"name": [
{
"language_code": "en_us",
"string": "Parent Network MAC"
}
]
},
{
"column": "Watched_Value3",
"mapped_to_column": "cur_PORT",
"css_classes": "col-sm-2",
"show": true,
"type": "label",
"default_value": "",
"options": [],
"localized": ["name"],
"name": [
{
"language_code": "en_us",
"string": "Port"
}
]
},
{
"column": "Watched_Value4",
"mapped_to_column": "cur_SSID",
"css_classes": "col-sm-2",
"show": true,
"type": "label",
"default_value": "",
"options": [],
"localized": ["name"],
"name": [
{
"language_code": "en_us",
"string": "SSID"
}
]
},
{
"column": "Extra",
"mapped_to_column": "cur_Type",
"css_classes": "col-sm-2",
"show": false,
"type": "label",
"default_value": "",
"options": [],
"localized": ["name"],
"name": [
{
"language_code": "en_us",
"string": "Site or Vendor"
}
]
},
{
"column": "Dummy",
"mapped_to_column": "cur_ScanMethod",
"mapped_to_column_data": {
"value": "OMDSDN"
},
"css_classes": "col-sm-2",
"show": true,
"type": "label",
"default_value": "",
"options": [],
"localized": ["name"],
"name": [
{
"language_code": "en_us",
"string": "Scan method"
},
{
"language_code": "es_es",
"string": "Método de escaneo"
},
{
"language_code": "de_de",
"string": "Scanmethode"
}
]
},
{
"column": "DateTimeCreated",
"css_classes": "col-sm-2",
"show": true,
"type": "label",
"default_value": "",
"options": [],
"localized": ["name"],
"name": [
{
"language_code": "en_us",
"string": "Created"
},
{
"language_code": "es_es",
"string": "Creado"
},
{
"language_code": "de_de",
"string": "Erstellt"
}
]
},
{
"column": "DateTimeChanged",
"css_classes": "col-sm-2",
"show": true,
"type": "label",
"default_value": "",
"options": [],
"localized": ["name"],
"name": [
{
"language_code": "en_us",
"string": "Changed"
},
{
"language_code": "es_es",
"string": "Cambiado"
},
{
"language_code": "de_de",
"string": "Geändert"
}
]
},
{
"column": "Status",
"css_classes": "col-sm-1",
"show": true,
"type": "replace",
"default_value": "",
"options": [
{
"equals": "watched-not-changed",
"replacement": "<div style='text-align:center'><i class='fa-solid fa-square-check'></i><div></div>"
},
{
"equals": "watched-changed",
"replacement": "<div style='text-align:center'><i class='fa-solid fa-triangle-exclamation'></i></div>"
},
{
"equals": "new",
"replacement": "<div style='text-align:center'><i class='fa-solid fa-circle-plus'></i></div>"
},
{
"equals": "missing-in-last-scan",
"replacement": "<div style='text-align:center'><i class='fa-solid fa-question'></i></div>"
}
],
"localized": ["name"],
"name": [
{
"language_code": "en_us",
"string": "Status"
},
{
"language_code": "es_es",
"string": "Estado"
},
{
"language_code": "de_de",
"string": "Status"
}
]
}
]
}

BIN
front/plugins/omada_sdn_imp/omada_account_sample.png Normal file
View File

Binary file not shown.
Side by Side

After

Width:  |  Height:  |  Size: 107 KiB

411
front/plugins/omada_sdn_imp/omada_sdn.py
View File

@@ -1,12 +1,33 @@
#!/usr/bin/env python
__author__ = "ffsb"
__version__ = "0.1" #initial
__version__ = "0.2" # added logic to retry omada api call once as it seems to sometimes fail for some reasons, and error handling logic...
__version__ = "0.3" # split devices API calls to allow multithreading but had to stop due to concurency issues.
__version__ = "0.6" # found issue with multithreading - my omada calls redirect stdout which gets clubbered by normal stdout... not sure how to fix for now...
# query OMADA SDN to populate NetAlertX witch omada switches, access points, clients.
# try to identify and populate their connections by switch/accesspoints and ports/SSID
# try to differentiate root bridges from accessory
#
# sample code to update unbound on opnsense - for reference...
# curl -X POST -d '{"host":{"enabled":"1","hostname":"test","domain":"testdomain.com","rr":"A","mxprio":"","mx":"","server":"10.0.1.1","description":""}}' -H "Content-Type: application/json" -k -u $OPNS_KEY:$OPNS_SECRET https://$IPFW/api/unbound/settings/AddHostOverride
#
import os
import pathlib
import sys
import json
import sqlite3
import tplink_omada_client
import importlib.util
import time
import io
import re
import concurrent.futures
#import netifaces
# Define the installation path and extend the system path for plugin imports
INSTALL_PATH = "/app"
sys.path.extend([f"{INSTALL_PATH}/front/plugins", f"{INSTALL_PATH}/server"])
@@ -22,26 +43,168 @@ from notification import write_notification
CUR_PATH = str(pathlib.Path(__file__).parent.resolve())
LOG_FILE = os.path.join(CUR_PATH, 'script.log')
RESULT_FILE = os.path.join(CUR_PATH, 'last_result.log')
OMADA_API_RETURN_FILE = os.path.join(CUR_PATH, 'omada_api_return')
# Initialize the Plugin obj output file
plugin_objects = Plugin_Objects(RESULT_FILE)
#
# sample target output:
# 0 MAC, 1 IP, 2 Name, 3 switch/AP, 4 port/SSID, 5 TYPE
#17:27:10 [<unique_prefix>] token: "['1A-2B-3C-4D-5E-6F', '192.168.0.217', '1A-2B-3C-4D-5E-6F', '17', '40-AE-30-A5-A7-50, 'Switch']"
# Constants for array indices
MAC, IP, NAME, SWITCH_AP, PORT_SSID, TYPE = range(6)
# sample omada devices input format:
#
# 0.MAC 1.IP 2.type 3.status 4.name 5.model
#40-AE-30-A5-A7-50 192.168.0.11 ap CONNECTED office_Access_point EAP773(US) v1.0
#B0-95-75-46-0C-39 192.168.0.4 switch CONNECTED pantry12 T1600G-52PS v4.0
dMAC, dIP, dTYPE, dSTATUS, dNAME, dMODEL = range(6)
# sample omada clients input format:
# 0 MAC, 1 IP, 2 Name, 3 switch/AP, 4 port/SSID,
#17:27:10 [<unique_prefix>] token: "['1A-2B-3C-4D-5E-6F', '192.168.0.217', '1A-2B-3C-4D-5E-6F', 'myssid_name2', '(office_Access_point)']"
#17:27:10 [<unique_prefix>] token: "['1A-2B-3C-4D-5E-01', '192.168.0.153', 'frontyard_ESP_29E753', 'pantry12', '(48)']"
#17:27:10 [<unique_prefix>] token: "['1A-2B-3C-4D-5E-02', '192.168.0.1', 'bastion', 'office24', '(23)']"
#17:27:10 [<unique_prefix>] token: "['1A-2B-3C-4D-5E-03', '192.168.0.226', 'brick', 'myssid_name3', '(office_Access_point)']"
cMAC, cIP, cNAME, cSWITCH_AP, cPORT_SSID = range(5)
OMDLOGLEVEL = 'debug'
pluginName = 'OMDSDN'
#
# translate MAC address from standard ieee model to ietf draft
# AA-BB-CC-DD-EE-FF to aa:bb:cc:dd:ee:ff
# tplink adheres to ieee, Nax adheres to ietf
def ieee2ietf_mac_formater(inputmac):
return(inputmac.lower().replace('-',':'))
def ietf2ieee_mac_formater(inputmac):
return(inputmac.upper().replace(':','-'))
def get_mac_from_IP(target_IP):
from scapy.all import ARP, Ether, srp
try:
arp_request = ARP(pdst=target_IP)
ether = Ether(dst="ff:ff:ff:ff:ff:ff")
packet = ether/arp_request
result = srp(packet, timeout=3, verbose=0)[0]
if result:
return result[0][1].hwsrc
else:
return None
except Exception as e:
mylog('minimal', [f'[{pluginName}] get_mac_from_IP ERROR:{e}'])
return None
#
# wrapper to call the omada python library's own wrapper
# it returns the output as a multiline python string
#
def callomada(myargs):
arguments=" ".join(myargs)
mylog('verbose', [f'[{pluginName}] callomada:{arguments}'])
from tplink_omada_client.cli import main as omada
from contextlib import redirect_stdout
omada_output = ''
retries = 2
while omada_output == '' and retries > 1:
retries = retries - 1
try:
mf = io.StringIO()
with redirect_stdout(mf):
bar = omada(myargs)
omada_output = mf.getvalue()
except Exception as e:
mylog('minimal', [f'[{pluginName}] ERROR WHILE CALLING callomada:{arguments}\n {mf}'])
omada_output= ''
return(omada_output)
#
# extract all the mac addresses from a multilines text...
# return a list of MAC as 'string'
#
def extract_mac_addresses(text):
mac_pattern = r"([0-9A-Fa-f]{2}[:-][0-9A-Fa-f]{2}[:-][0-9A-Fa-f]{2}[:-][0-9A-Fa-f]{2}[:-][0-9A-Fa-f]{2}[:-][0-9A-Fa-f]{2})"
mac_addresses = re.findall(mac_pattern, text)
return ["".join(parts) for parts in mac_addresses]
def find_default_gateway_ip ():
#import netifaces
#gw = netifaces.gateways()
#return(gw['default'][netifaces.AF_INET][0])
from scapy.all import conf, Route, sr1, IP, ICMP
default_route = conf.route.route("0.0.0.0")
return default_route[2] if default_route[2] else None
#return('192.168.0.1')
def add_uplink (uplink_mac, switch_mac, device_data_bymac, sadevices_linksbymac,port_byswitchmac_byclientmac):
#mylog(OMDLOGLEVEL, [f'[{pluginName}] trying to add uplink="{uplink_mac}" to switch="{switch_mac}"'])
#mylog(OMDLOGLEVEL, [f'[{pluginName}] before adding:"{device_data_bymac[switch_mac]}"'])
if device_data_bymac[switch_mac][SWITCH_AP] == 'null':
device_data_bymac[switch_mac][SWITCH_AP] = uplink_mac
if device_data_bymac[switch_mac][TYPE] == 'Switch' and device_data_bymac[uplink_mac][TYPE] == 'Switch':
port_to_uplink = port_byswitchmac_byclientmac[switch_mac][uplink_mac]
#find_port_of_uplink_switch(switch_mac, uplink_mac)
else:
port_to_uplink=device_data_bymac[uplink_mac][PORT_SSID]
device_data_bymac[switch_mac][PORT_SSID] = port_to_uplink
# mylog(OMDLOGLEVEL, [f'[{pluginName}] after adding:"{device_data_bymac[switch_mac]}"'])
for link in sadevices_linksbymac[switch_mac]:
if device_data_bymac[link][SWITCH_AP] == 'null' and device_data_bymac[switch_mac][TYPE] == 'Switch':
add_uplink(switch_mac, link, device_data_bymac, sadevices_linksbymac,port_byswitchmac_byclientmac)
pluginName = '<unique_prefix>'
# ----------------------------------------------
# Main initialization
def main():
mylog('verbose', [f'[{pluginName}] In script'])
start_time = time.time()
mylog('verbose', [f'[{pluginName}] starting execution'])
from database import DB
from device import Device_obj
db = DB() # instance of class DB
db.open()
# Create a Device_obj instance
device_handler = Device_obj(db)
# Retrieve configuration settings
some_setting = get_setting_value('OMDSDN_url')
# these should be self-explanatory
omada_sites = []
omada_username = get_setting_value('OMDSDN_username')
omada_password = get_setting_value('OMDSDN_password')
omada_sites = get_setting_value('OMDSDN_sites')
omada_site = omada_sites[0]
omada_url = get_setting_value('OMDSDN_url')
omada_login = callomada(['-t','myomada','target','--url',omada_url,'--user',omada_username,
'--password',omada_password,'--site',omada_site,'--set-default'])
mylog('verbose', [f'[{pluginName}] login to omada result is: {omada_login}'])
clients_list = callomada(['-t','myomada','clients'])
mylog('verbose', [f'[{pluginName}] clients found:"{clients_list.count("\n")}"\n{clients_list}'])
switches_and_aps = callomada(['-t','myomada','devices'])
mylog('verbose', [f'[{pluginName}] omada devices (switches, access points) found:"{switches_and_aps.count("\n")}" \n {switches_and_aps}'])
#some_setting = get_setting_value('OMDSDN_url')
#mylog(OMDLOGLEVEL, [f'[{pluginName}] some_setting value {some_setting}'])
mylog(OMDLOGLEVEL, [f'[{pluginName}] ffsb'])
mylog('verbose', [f'[{pluginName}] some_setting calue {some_setting}'])
# retrieve data
device_data = get_device_data(some_setting)
device_data = get_device_data(clients_list, switches_and_aps, device_handler)
# Process the data into native application tables
mylog('verbose', [f'[{pluginName}] New entries to create: "{len(device_data)}"'])
if len(device_data) > 0:
# insert devices into the lats_result.log
@@ -50,36 +213,238 @@ def main():
# {
# "column": "Object_PrimaryID", <--------- the value I save into primaryId
# "mapped_to_column": "cur_MAC", <--------- gets unserted into the CurrentScan DB table column cur_MAC
#
for device in device_data:
plugin_objects.add_object(
primaryId = device['some_id'], # MAC
secondaryId = device['some_id'], # IP
watched1 = device['some_id'], # NAME/HOSTNAME
watched2 = device['some_id'], # PARENT NETWORK NODE MAC
watched3 = device['some_id'], # PORT
watched4 = device['some_id'], # SSID
extra = device['some_id'], # SITENAME (cur_NetworkSite) or VENDOR (cur_Vendor) (PICK one and adjust config.json -> "column": "Extra")
foreignKey = device['some_id']) # usually MAC
# watched1 = 'null' ,
# figure a way to run my udpate script delayed
mylog('verbose', [f'[{pluginName}] New entries: "{len(new_devices)}"'])
for device in device_data:
mylog(OMDLOGLEVEL, [f'[{pluginName}] main parsing device: "{device}"'])
myport = device[PORT_SSID] if device[PORT_SSID].isdigit() else ''
myssid = device[PORT_SSID] if not device[PORT_SSID].isdigit() else ''
ParentNetworkNode = ieee2ietf_mac_formater(device[SWITCH_AP]) if device[SWITCH_AP] != 'Internet' else 'Internet'
mymac = ieee2ietf_mac_formater(device[MAC])
plugin_objects.add_object(
primaryId = mymac, # MAC
secondaryId = device[IP], # IP
watched1 = device[NAME], # NAME/HOSTNAME
watched2 = ParentNetworkNode, # PARENT NETWORK NODE MAC
watched3 = myport, # PORT
watched4 = myssid, # SSID
extra = device[TYPE],
#omada_site, # SITENAME (cur_NetworkSite) or VENDOR (cur_Vendor) (PICK one and adjust config.json -> "column": "Extra")
foreignKey = device[MAC].lower().replace('-',':')) # usually MAC
mylog('verbose', [f'[{pluginName}] New entries: "{mymac:<18}, {device[IP]:<16}, {device[NAME]:<63}, {ParentNetworkNode:<18}, {myport:<4}, {myssid:<32}, {device[TYPE]}"'])
mylog('verbose', [f'[{pluginName}] New entries: "{len(device_data)}"'])
# log result
# plugin_objects.write_result_file()
plugin_objects.write_result_file()
#mylog(OMDLOGLEVEL, [f'[{pluginName}] TEST name from MAC: {device_handler.getValueWithMac('dev_Name','00:e2:59:00:a0:8e')}'])
#mylog(OMDLOGLEVEL, [f'[{pluginName}] TEST MAC from IP: {get_mac_from_IP('192.168.0.1')} also {ietf2ieee_mac_formater(get_mac_from_IP('192.168.0.1'))}'])
end_time = time.time()
mylog('verbose', [f'[{pluginName}] execution completed in {end_time - start_time:.2f} seconds'])
return 0
def get_omada_devices_details(msadevice_data):
mthisswitch = msadevice_data[dMAC]
mtype = msadevice_data[dTYPE]
mswitch_detail = ''
mswitch_dump = ''
if mtype == 'ap':
mswitch_detail = callomada(['access-point', mthisswitch])
elif mtype == 'switch':
mswitch_detail = callomada(['switch', mthisswitch])
mswitch_dump = callomada(['-t','myomada','switch','-d',mthisswitch])
else:
mswitch_detail = ''
nswitch_dump = ''
return mswitch_detail, mswitch_dump
# ----------------------------------------------
# retrieve data
def get_device_data(some_setting):
def get_device_data(omada_clients_output,switches_and_aps,device_handler):
device_data = []
# sample omada devices input format:
# 0.MAC 1.IP 2.type 3.status 4.name 5.model
#40-AE-30-A5-A7-50 192.168.0.11 ap CONNECTED office_Access_point EAP773(US) v1.0
#B0-95-75-46-0C-39 192.168.0.4 switch CONNECTED pantry12 T1600G-52PS v4.0
#
# sample target output:
# 0 MAC, 1 IP, 2 Name, 3 switch/AP, 4 port/SSID, 5 TYPE
#17:27:10 [<unique_prefix>] token: "['1A-2B-3C-4D-5E-6F', '192.168.0.217', '1A-2B-3C-4D-5E-6F', '17', '40-AE-30-A5-A7-50, 'Switch']"
#constants
sadevices_macbyname = {}
sadevices_macbymac = {}
sadevices_linksbymac = {}
port_byswitchmac_byclientmac = {}
device_data_bymac = {}
device_data_mac_byip = {}
omada_force_overwrite = get_setting_value('OMDSDN_force_overwrite')
switch_details = {}
switch_dumps = {}
sadevices = switches_and_aps.splitlines()
mylog(OMDLOGLEVEL, [f'[{pluginName}] switches_and_aps rows: "{len(sadevices)}"'])
for sadevice in sadevices:
sadevice_data = sadevice.split()
thisswitch = sadevice_data[dMAC]
thistype = sadevice_data[dTYPE]
switch_details[thisswitch], switch_dumps[thisswitch] = get_omada_devices_details(sadevice_data)
mylog('verbose', [f'[{pluginName}] switches details collected "{len(switch_details)}"'])
mylog('verbose', [f'[{pluginName}] dump details collected "{len(switch_details)}"'])
# Using ThreadPoolExecutor for parallel execution
for sadevice in sadevices:
sadevice_data = sadevice.split()
thisswitch = sadevice_data[dMAC]
sadevices_macbyname[sadevice_data[4]] = thisswitch
if sadevice_data[dTYPE] == 'ap':
sadevice_type = 'AP'
#sadevice_details = callomada(['access-point', thisswitch])
sadevice_details = switch_details[thisswitch]
if sadevice_details == '':
sadevice_links = [thisswitch]
else:
sadevice_links = extract_mac_addresses(sadevice_details)
sadevices_linksbymac[thisswitch] = sadevice_links[1:]
#mylog(OMDLOGLEVEL, [f'[{pluginName}]adding switch details: "{sadevice_details}"'])
#mylog(OMDLOGLEVEL, [f'[{pluginName}]links are: "{sadevice_links}"'])
#mylog(OMDLOGLEVEL, [f'[{pluginName}]linksbymac are: "{sadevices_linksbymac[thisswitch]}"'])
elif sadevice_data[dTYPE] == 'switch':
sadevice_type = 'Switch'
#sadevice_details=callomada(['switch', thisswitch])
sadevice_details = switch_details[thisswitch]
if sadevice_details == '':
sadevice_links = [thisswitch]
else:
sadevice_links=extract_mac_addresses(sadevice_details)
sadevices_linksbymac[thisswitch] = sadevice_links[1:]
# recovering the list of switches connected to sadevice switch and on which port...
#switchdump = callomada(['-t','myomada','switch','-d',thisswitch])
switchdump = switch_dumps[thisswitch]
mylog(OMDLOGLEVEL, [f'[{pluginName}] switchdump: {switchdump}'])
port_byswitchmac_byclientmac[thisswitch] = {}
for link in sadevices_linksbymac[thisswitch]:
port_pattern = r"(?:{[^}]*\"port\"\: )([0-9]+)(?=[^}]*"+re.escape(link)+r")"
myport = re.findall(port_pattern, switchdump,re.DOTALL)
#mylog(OMDLOGLEVEL, [f'[{pluginName}] switchdump: link={link} myport:{myport}'])
port_byswitchmac_byclientmac[thisswitch][link] = myport[0] if myport else ''
#mylog(OMDLOGLEVEL, [f'[{pluginName}]links are: "{sadevice_links}"'])
#mylog(OMDLOGLEVEL, [f'[{pluginName}]linksbymac are: "{sadevices_linksbymac[thisswitch]}"'])
#mylog(OMDLOGLEVEL, [f'[{pluginName}]ports of each links are: "{port_byswitchmac_byclientmac[thisswitch]}"'])
#mylog(OMDLOGLEVEL, [f'[{pluginName}]adding switch details: "{sadevice_details}"'])
else:
sadevice_type = 'null'
sadevice_details='null'
device_data_bymac[thisswitch] = [thisswitch, sadevice_data[dIP], sadevice_data[dNAME], 'null', 'null',sadevice_type]
device_data_mac_byip[sadevice_data[dIP]] = thisswitch
foo=[thisswitch, sadevice_data[1], sadevice_data[4], 'null', 'null']
mylog(OMDLOGLEVEL, [f'[{pluginName}]adding switch: "{foo}"'])
# sadevices_macbymac[thisswitch] = thisswitch
mylog(OMDLOGLEVEL, [f'[{pluginName}] switch_macbyname: "{sadevices_macbyname}"'])
mylog(OMDLOGLEVEL, [f'[{pluginName}] switches: "{device_data_bymac}"'])
# do some processing, call exteranl APIs, and return a device list
# ...
#
# sample omada clients input format:
# 0 MAC, 1 IP, 2 Name, 3 switch/AP, 4 port/SSID,
#17:27:10 [<unique_prefix>] token: "['1A-2B-3C-4D-5E-6F', '192.168.0.217', '1A-2B-3C-4D-5E-6F', 'myssid_name2', '(office_Access_point)']"
#17:27:10 [<unique_prefix>] token: "['1A-2B-3C-4D-5E-01', '192.168.0.153', 'frontyard_ESP_29E753', 'pantry12', '(48)']"
#17:27:10 [<unique_prefix>] token: "['1A-2B-3C-4D-5E-02', '192.168.0.1', 'bastion', 'office24', '(23)']"
#17:27:10 [<unique_prefix>] token: "['1A-2B-3C-4D-5E-03', '192.168.0.226', 'brick', 'myssid_name3', '(office_Access_point)']"
# sample target output:
# 0 MAC, 1 IP, 2 Name, 3 MAC of switch/AP, 4 port/SSID, 5 TYPE
#17:27:10 [<unique_prefix>] token: "['1A-2B-3C-4D-5E-6F', '192.168.0.217', 'brick', 'office_Access_point','myssid_name2', , 'Switch']"
return device_data
odevices = omada_clients_output.splitlines()
mylog(OMDLOGLEVEL, [f'[{pluginName}] omada_clients_outputs rows: "{len(odevices)}"'])
for odevice in odevices:
odevice_data = odevice.split()
odevice_data_reordered = [ MAC, IP, NAME, SWITCH_AP, PORT_SSID, TYPE]
odevice_data_reordered[MAC]=odevice_data[cMAC]
odevice_data_reordered[IP]=odevice_data[cIP]
real_naxname = device_handler.getValueWithMac('dev_Name',ieee2ietf_mac_formater(odevice_data[cMAC]))
#
# if the name stored in Nax for a device is empty or the MAC addres or has some parenthhesis or is the same as in omada
# don't bother updating omada's name at all.
#
naxname = real_naxname
if real_naxname != None:
if '(' in real_naxname:
# removing parenthesis and domains from the name
naxname = real_naxname.split('(')[0]
if naxname != None and '.' in naxname:
naxname = naxname.split('.')[0]
if naxname in ( None, 'null', '' ):
naxname = odevice_data[cNAME] if odevice_data[cNAME] != '' else odevice_data[cMAC]
naxname = naxname.strip()
mylog('debug', [f'[{pluginName}] TEST name from MAC: {naxname}'])
if odevice_data[cNAME] in (odevice_data[cMAC], 'null', ''):
mylog('verbose', [f'[{pluginName}] updating omada server because odevice_data is: {odevice_data[cNAME]} and naxname is: "{naxname}"'])
callomada(['set-client-name', odevice_data[cMAC], naxname])
odevice_data_reordered[NAME] = naxname
else:
if omada_force_overwrite and naxname != odevice_data[cNAME] :
mylog('verbose', [f'[{pluginName}] updating omada server because odevice_data is: "{odevice_data[cNAME]} and naxname is: "{naxname}"'])
callomada(['set-client-name', odevice_data[cMAC], naxname])
odevice_data_reordered[NAME] = naxname
mightbeport = odevice_data[cPORT_SSID].lstrip('(')
mightbeport = mightbeport.rstrip(')')
if mightbeport.isdigit():
odevice_data_reordered[SWITCH_AP] = odevice_data[cSWITCH_AP]
odevice_data_reordered[PORT_SSID] = mightbeport
else:
odevice_data_reordered[SWITCH_AP] = mightbeport
odevice_data_reordered[PORT_SSID] = odevice_data[cSWITCH_AP]
# replacing the switch name with its MAC...
try:
mightbemac = sadevices_macbyname[odevice_data_reordered[SWITCH_AP]]
odevice_data_reordered[SWITCH_AP] = mightbemac
except KeyError:
mylog(OMDLOGLEVEL, [f'[{pluginName}] could not find the mac adddress for: "{odevice_data_reordered[SWITCH_AP]}"'])
# adding the type
odevice_data_reordered[TYPE] = 'null'
device_data_bymac[odevice_data_reordered[MAC]] = odevice_data_reordered
device_data_mac_byip[odevice_data_reordered[IP]] = odevice_data_reordered[MAC]
mylog(OMDLOGLEVEL, [f'[{pluginName}] tokens: "{odevice_data}"'])
mylog(OMDLOGLEVEL, [f'[{pluginName}] tokens_reordered: "{odevice_data_reordered}"'])
# populating the uplinks nodes of the omada switches and access points manually
# since OMADA SDN makes is unreliable if the gateway is not their own tplink hardware...
# step1 let's find the the default router
#
default_router_ip = find_default_gateway_ip()
default_router_mac = ietf2ieee_mac_formater(get_mac_from_IP(default_router_ip))
device_data_bymac[default_router_mac][TYPE] = 'Firewall'
# step2 let's find the first switch and set the default router parent to internet
first_switch=device_data_bymac[default_router_mac][SWITCH_AP]
device_data_bymac[default_router_mac][SWITCH_AP] = 'Internet'
# step3 let's set the switch connected to the default gateway uplink to the default gateway and hardcode port to 1 for now:
#device_data_bymac[first_switch][SWITCH_AP]=default_router_mac
#device_data_bymac[first_switch][SWITCH_AP][PORT_SSID] = '1'
# step4, let's go recursively through switches other links to mark update their uplinks
# and pray it ends one day...
#
add_uplink(default_router_mac,first_switch, device_data_bymac,sadevices_linksbymac,port_byswitchmac_byclientmac)
return device_data_bymac.values()
if __name__ == '__main__':
main()

498
front/plugins/omada_sdn_imp/omada_sdn.py.0.3 Executable file
View File

@@ -0,0 +1,498 @@
#!/usr/bin/env python
"""
Omada SDN Query Script
This script queries the OMADA SDN to populate NetAlertX with Omada switches, access points, and clients.
It attempts to identify and populate their connections by switch/access points and ports/SSID,
and tries to differentiate root bridges from accessories.
Author: ffsb
Version: 0.2 - Added logic to retry Omada API call once as it sometimes fails, and improved error handling.
"""
__author__ = "ffsb"
__version__ = "0.1" #initial
__version__ = "0.2" # added logic to retry omada api call once as it seems to sometimes fail for some reasons, and error handling logic...
__version__ = "0.3" # adding parallelism
#
# sample code to update unbound on opnsense - for reference...
# curl -X POST -d '{"host":{"enabled":"1","hostname":"test","domain":"testdomain.com","rr":"A","mxprio":"","mx":"","server":"10.0.1.1","description":""}}' -H "Content-Type: application/json" -k -u $OPNS_KEY:$OPNS_SECRET https://$IPFW/api/unbound/settings/AddHostOverride
#
import os
import pathlib
import sys
import json
import sqlite3
import tplink_omada_client
import importlib.util
import time
import io
import re
import concurrent.futures
from queue import Queue
import multiprocessing
from multiprocessing import Pool, Manager
import os
# Define the installation path and extend the system path for plugin imports
INSTALL_PATH = "/app"
sys.path.extend([f"{INSTALL_PATH}/front/plugins", f"{INSTALL_PATH}/server"])
from plugin_helper import Plugin_Object, Plugin_Objects, decodeBase64
from plugin_utils import get_plugins_configs
from logger import mylog
from const import pluginsPath, fullDbPath
from helper import timeNowTZ, get_setting_value
from notification import write_notification
# Define the current path and log file paths
CUR_PATH = str(pathlib.Path(__file__).parent.resolve())
LOG_FILE = os.path.join(CUR_PATH, 'script.log')
RESULT_FILE = os.path.join(CUR_PATH, 'last_result.log')
# Initialize the Plugin obj output file
plugin_objects = Plugin_Objects(RESULT_FILE)
pluginName = 'OMDSDN'
#
# sample target output:
# 0 MAC, 1 IP, 2 Name, 3 switch/AP, 4 port/SSID, 5 TYPE
#17:27:10 [<unique_prefix>] token: "['9C-04-A0-82-67-45', '192.168.0.217', '9C-04-A0-82-67-45', '17', '40-AE-30-A5-A7-50, 'Switch']"
# Constants for array indices
MAC, IP, NAME, SWITCH_AP, PORT_SSID, TYPE = range(6)
# sample omada devices input format:
#
# 0.MAC 1.IP 2.type 3.status 4.name 5.model
#40-AE-30-A5-A7-50 192.168.0.11 ap CONNECTED ompapaoffice EAP773(US) v1.0
#B0-95-75-46-0C-39 192.168.0.4 switch CONNECTED pantry12 T1600G-52PS v4.0
dMAC, dIP, dTYPE, dSTATUS, dNAME, dMODEL = range(6)
# sample omada clients input format:
# 0 MAC, 1 IP, 2 Name, 3 switch/AP, 4 port/SSID,
#17:27:10 [<unique_prefix>] token: "['9C-04-A0-82-67-45', '192.168.0.217', '9C-04-A0-82-67-45', 'froggies2', '(ompapaoffice)']"
#17:27:10 [<unique_prefix>] token: "['50-02-91-29-E7-53', '192.168.0.153', 'frontyard_ESP_29E753', 'pantry12', '(48)']"
#17:27:10 [<unique_prefix>] token: "['00-E2-59-00-A0-8E', '192.168.0.1', 'bastion', 'office24', '(23)']"
#17:27:10 [<unique_prefix>] token: "['60-DD-8E-CA-A4-B3', '192.168.0.226', 'brick', 'froggies3', '(ompapaoffice)']"
cMAC, cIP, cNAME, cSWITCH_AP, cPORT_SSID = range(5)
OMDLOGLEVEL = 'verbose'
def ieee2ietf_mac_formater(inputmac):
"""Translate MAC address from standard IEEE model to IETF draft."""
return inputmac.lower().replace('-', ':')
def ietf2ieee_mac_formater(inputmac):
"""Translate MAC address from IETF draft to standard IEEE model."""
return inputmac.upper().replace(':', '-')
def get_mac_from_IP(target_IP):
"""Get MAC address from IP using ARP."""
from scapy.all import ARP, Ether, srp
try:
arp_request = ARP(pdst=target_IP)
ether = Ether(dst="ff:ff:ff:ff:ff:ff")
packet = ether/arp_request
result = srp(packet, timeout=3, verbose=0)[0]
if result:
return result[0][1].hwsrc
else:
return None
except Exception as e:
mylog('minimal', [f'[{pluginName}] get_mac_from_IP ERROR:{e}'])
return None
def callomada(myargs):
"""Wrapper to call the Omada python library's own wrapper."""
arguments = " ".join(myargs)
mylog('verbose', [f'[{pluginName}] callomada START:{arguments}'])
from tplink_omada_client.cli import main as omada
from contextlib import redirect_stdout
omada_output = ''
retries = 2
while omada_output == '' and retries > 0:
retries -= 1
try:
mf = io.StringIO()
with redirect_stdout(mf):
omada(myargs)
omada_output = mf.getvalue()
except Exception as e:
mylog('minimal', [f'[{pluginName}] ERROR WHILE CALLING callomada:{arguments}\n {e}'])
omada_output = ''
mylog('verbose', [f'[{pluginName}] callomada END:{arguments}'])
return omada_output
def extract_mac_addresses(text):
"""Extract all the MAC addresses from multiline text."""
mac_pattern = r"([0-9A-Fa-f]{2}[:-][0-9A-Fa-f]{2}[:-][0-9A-Fa-f]{2}[:-][0-9A-Fa-f]{2}[:-][0-9A-Fa-f]{2}[:-][0-9A-Fa-f]{2})"
return re.findall(mac_pattern, text)
def find_default_gateway_ip():
"""Find the default gateway IP address."""
from scapy.all import conf, Route
default_route = conf.route.route("0.0.0.0")
return default_route[2] if default_route[2] else None
def add_uplink(uplink_mac, switch_mac, device_data_bymac, sadevices_linksbymac, port_byswitchmac_byclientmac):
"""Add uplink information to switches recursively."""
mylog(OMDLOGLEVEL, [f'[{pluginName}] trying to add uplink="{uplink_mac}" to switch="{switch_mac}"'])
mylog(OMDLOGLEVEL, [f'[{pluginName}] before adding:"{device_data_bymac[switch_mac]}"'])
mylog(OMDLOGLEVEL, [f'[{pluginName}] this are the port_byswitchmac:"{port_byswitchmac_byclientmac[switch_mac]}"'])
if device_data_bymac[switch_mac][SWITCH_AP] == 'null':
device_data_bymac[switch_mac][SWITCH_AP] = uplink_mac
if device_data_bymac[switch_mac][TYPE] == 'Switch' and device_data_bymac[uplink_mac][TYPE] == 'Switch':
port_to_uplink = port_byswitchmac_byclientmac[switch_mac][uplink_mac]
else:
port_to_uplink = device_data_bymac[uplink_mac][PORT_SSID]
device_data_bymac[switch_mac][PORT_SSID] = port_to_uplink
mylog(OMDLOGLEVEL, [f'[{pluginName}] after adding:"{device_data_bymac[switch_mac]}"'])
for link in sadevices_linksbymac[switch_mac]:
if device_data_bymac[link][SWITCH_AP] == 'null' and device_data_bymac[switch_mac][TYPE] == 'Switch':
add_uplink(switch_mac, link, device_data_bymac, sadevices_linksbymac, port_byswitchmac_byclientmac)
def main():
"""Main function to execute the script."""
start_time = time.time()
mylog('verbose', [f'[{pluginName}] starting execution'])
from database import DB
from device import Device_obj
db = DB()
db.open()
device_handler = Device_obj(db)
# Retrieve configuration settings
omada_username = get_setting_value('OMDSDN_username')
omada_password = get_setting_value('OMDSDN_password')
omada_sites = get_setting_value('OMDSDN_sites')
omada_site = omada_sites[0]
omada_url = get_setting_value('OMDSDN_url')
# Login to Omada
omada_login = callomada(['-t', 'myomada', 'target', '--url', omada_url, '--user', omada_username,
'--password', omada_password, '--site', omada_site, '--set-default'])
mylog('verbose', [f'[{pluginName}] login to omada result is: {omada_login}'])
# Get clients and devices
clients_list = callomada(['-t', 'myomada', 'clients'])
mylog('verbose', [f'[{pluginName}] clients found:"{clients_list.count("\n")}"\n{clients_list}'])
switches_and_aps = callomada(['-t', 'myomada', 'devices'])
mylog('verbose', [f'[{pluginName}] omada devices (switches, access points) found:"{switches_and_aps.count("\n")}" \n {switches_and_aps}'])
# Process data
device_data = get_device_data(clients_list, switches_and_aps, device_handler)
mylog('verbose', [f'[{pluginName}] New entries to create: "{len(device_data)}"'])
if len(device_data) > 0:
for device in device_data:
mylog(OMDLOGLEVEL, [f'[{pluginName}] main parsing device: "{device}"'])
myport = device[PORT_SSID] if device[PORT_SSID].isdigit() else ''
myssid = device[PORT_SSID] if not device[PORT_SSID].isdigit() else ''
ParentNetworkNode = ieee2ietf_mac_formater(device[SWITCH_AP]) if device[SWITCH_AP] != 'Internet' else 'Internet'
plugin_objects.add_object(
primaryId = ieee2ietf_mac_formater(device[MAC]),
secondaryId = device[IP],
watched1 = device[NAME] if device[NAME] != 'null' else '',
watched2 = ParentNetworkNode,
watched3 = myport,
watched4 = myssid,
extra = device[TYPE] if device[TYPE] != 'null' else '',
foreignKey = ieee2ietf_mac_formater(device[MAC])
)
mylog(OMDLOGLEVEL, [f'[{pluginName}] New entries: "{len(device_data)}"'])
# Write results
plugin_objects.write_result_file()
end_time = time.time()
mylog('verbose', [f'[{pluginName}] execution completed in {end_time - start_time:.2f} seconds'])
return 0
'''
# version 0.3b
def get_omada_devices_details(sadevice_data,switch_details,switch_dumps):
"""Get device details from Omada. saved into a dictionary of strings"""
mylog(OMDLOGLEVEL, [f'[{pluginName}]getting the omada devices details: "{sadevice_data}"'])
thisswitch = sadevice_data[dMAC]
if sadevice_data[dTYPE] == 'ap':
switch_details[thisswitch] = callomada(['access-point', thisswitch])
elif sadevice_data[dTYPE] == 'switch':
switch_details[thisswitch] = callomada(['switch', thisswitch])
switch_dumps[thisswitch] = callomada(['-t','myomada','switch','-d',thisswitch])
else:
switch_details[thisswitch] = 'null'
switch_dumps[thisswitch] = 'null'
return
'''
'''
# version 0.3c
def get_omada_devices_details(sadevice_data):
mthisswitch = sadevice_data[dMAC]
mswitch_detail = ''
mswitch_dump = ''
if sadevice_data[dTYPE] == 'ap':
mswitch_detail = callomada(['access-point', mthisswitch])
elif sadevice_data[dTYPE] == 'switch':
mswitch_detail = callomada(['switch', mthisswitch])
mswitch_dump = callomada(['-t','myomada','switch','-d',mthisswitch])
else:
mswitch_detail = 'null'
nswitch_dump = 'null'
return mthisswitch, mswitch_detail, mswitch_dump
'''
def get_omada_devices_details(sadevice_data):
thisswitch = sadevice_data[dMAC]
try:
if sadevice_data[dTYPE] == 'ap':
switch_detail = callomada(['access-point', thisswitch])
return thisswitch, switch_detail, None
elif sadevice_data[dTYPE] == 'switch':
switch_detail = callomada(['switch', thisswitch])
switch_dump = callomada(['-t','myomada','switch','-d',thisswitch])
return thisswitch, switch_detail, switch_dump
else:
return thisswitch, 'null', 'null'
except Exception as e:
mylog('error', [f'[{pluginName}] Error processing {thisswitch}: {str(e)}'])
return thisswitch, 'error', 'error'
def get_device_data(omada_clients_output, switches_and_aps, device_handler):
"""Process and return device data from Omada output."""
"""
switch_dumps = {}
switch_details = {}
sadevices_macbyname = {}
sadevices_macbymac = {}
sadevices_linksbymac = {}
port_byswitchmac_byclientmac = {}
device_data_bymac = {}
device_data_mac_byip = {}
omada_force_overwrite = get_setting_value('OMDSDN_force_overwrite')
"""
manager = Manager()
switch_dumps = manager.dict()
switch_details = manager.dict()
sadevices_macbyname = manager.dict()
sadevices_macbymac = manager.dict()
sadevices_linksbymac = manager.dict()
port_byswitchmac_byclientmac = manager.dict()
device_data_bymac = manager.dict()
device_data_mac_byip = manager.dict()
omada_force_overwrite = get_setting_value('OMDSDN_force_overwrite')
sadevices = switches_and_aps.splitlines()
mylog(OMDLOGLEVEL, [f'[{pluginName}] switches_and_aps rows: "{len(sadevices)}"'])
'''
for sadevice in sadevices:
sadevice_data = sadevice.split()
get_omada_devices_details(sadevice_data,switch_details,switch_dumps)
'''
'''
# Create a ThreadPoolExecutor
# version 0.3b
with concurrent.futures.ThreadPoolExecutor(max_workers=4) as executor:
# Submit tasks for each device
futures = []
for sadevice in sadevices:
sadevice_data = sadevice.split()
future = executor.submit(get_omada_devices_details, sadevice_data, switch_details, switch_dumps)
futures.append(future)
# Wait for all tasks to complete
concurrent.futures.wait(futures)
'''
'''
# version 0.3c
# Create a ThreadPoolExecutor
with concurrent.futures.ThreadPoolExecutor(max_workers=3) as executor:
# Submit tasks for each device
future_to_device = {executor.submit(get_omada_devices_details, sadevice.split()): sadevice for sadevice in sadevices}
# Process results as they complete
for future in concurrent.futures.as_completed(future_to_device):
csadevice = future_to_device[future]
try:
mylog('verbose', [f'[{pluginName}] processing results of: {csadevice}'])
cthisswitch, cswitch_detail, cswitch_dump = future.result()
switch_details[cthisswitch] = cswitch_detail
switch_dumps[cthisswitch] = cswitch_dump
except Exception as exc:
mylog('error', [f'[{pluginName}] {csadevice} generated an exception: {exc}'])
'''
# Use multiprocessing Pool
with Pool(processes=3) as pool:
results = pool.map(get_omada_devices_details, [sadevice.split() for sadevice in sadevices])
mylog(OMDLOGLEVEL, [f'[{pluginName}] All API calls completed. Processing results...'])
# Process results
for thisswitch, switch_detail, switch_dump in results:
switch_details[thisswitch] = switch_detail
if switch_dump is not None:
switch_dumps[thisswitch] = switch_dump
mylog(OMDLOGLEVEL, [f'[{pluginName}] Finished collecting device details. Processing data...'])
# Now process the collected data
for sadevice in sadevices:
sadevice_data = sadevice.split()
thisswitch = sadevice_data[dMAC]
sadevices_macbyname[sadevice_data[4]] = thisswitch
if sadevice_data[dTYPE] == 'ap':
sadevice_type = 'AP'
#sadevice_details = callomada(['access-point', thisswitch])
sadevice_details = switch_details[thisswitch]
if sadevice_details == '':
sadevice_links = [thisswitch]
else:
sadevice_links = extract_mac_addresses(sadevice_details)
sadevices_linksbymac[thisswitch] = sadevice_links[1:]
mylog(OMDLOGLEVEL, [f'[{pluginName}]adding switch details: "{sadevice_details}"'])
mylog(OMDLOGLEVEL, [f'[{pluginName}]links are: "{sadevice_links}"'])
mylog(OMDLOGLEVEL, [f'[{pluginName}]linksbymac are: "{sadevices_linksbymac[thisswitch]}"'])
elif sadevice_data[dTYPE] == 'switch':
sadevice_type = 'Switch'
#sadevice_details=callomada(['switch', thisswitch])
sadevice_details = switch_details[thisswitch]
if sadevice_details == '':
sadevice_links = [thisswitch]
else:
sadevice_links=extract_mac_addresses(sadevice_details)
sadevices_linksbymac[thisswitch] = sadevice_links[1:]
# recovering the list of switches connected to sadevice switch and on which port...
#switchdump = callomada(['-t','myomada','switch','-d',thisswitch])
switchdump = switch_dumps[thisswitch]
port_byswitchmac_byclientmac[thisswitch] = {}
for link in sadevices_linksbymac[thisswitch]:
port_pattern = r"(?:{[^}]*\"port\"\: )([0-9]+)(?=[^}]*"+re.escape(link)+r")"
myport = re.findall(port_pattern, switchdump,re.DOTALL)
port_byswitchmac_byclientmac[thisswitch][link] = myport[0]
mylog(OMDLOGLEVEL, [f'[{pluginName}]links are: "{sadevice_links}"'])
mylog(OMDLOGLEVEL, [f'[{pluginName}]linksbymac are: "{sadevices_linksbymac[thisswitch]}"'])
mylog(OMDLOGLEVEL, [f'[{pluginName}]ports of each links are: "{port_byswitchmac_byclientmac[thisswitch]}"'])
mylog(OMDLOGLEVEL, [f'[{pluginName}]adding switch details: "{sadevice_details}"'])
else:
sadevice_type = 'null'
sadevice_details='null'
device_data_bymac[thisswitch] = [thisswitch, sadevice_data[dIP], sadevice_data[dNAME], 'null', 'null',sadevice_type]
device_data_mac_byip[sadevice_data[dIP]] = thisswitch
foo=[thisswitch, sadevice_data[1], sadevice_data[4], 'null', 'null']
mylog(OMDLOGLEVEL, [f'[{pluginName}]adding switch: "{foo}"'])
# sadevices_macbymac[thisswitch] = thisswitch
mylog(OMDLOGLEVEL, [f'[{pluginName}] switch_macbyname: "{sadevices_macbyname}"'])
mylog(OMDLOGLEVEL, [f'[{pluginName}] switches: "{device_data_bymac}"'])
# do some processing, call exteranl APIs, and return a device list
# ...
""" MAC = 0
IP = 1
NAME = 2
SWITCH_AP = 3
PORT_SSID = 4
TYPE = 5 """
# sample target output:
# 0 MAC, 1 IP, 2 Name, 3 MAC of switch/AP, 4 port/SSID, 5 TYPE
#17:27:10 [<unique_prefix>] token: "['9C-04-A0-82-67-45', '192.168.0.217', 'brick', 'ompapaoffice','froggies2', , 'Switch']"
odevices = omada_clients_output.splitlines()
mylog(OMDLOGLEVEL, [f'[{pluginName}] omada_clients_outputs rows: "{len(odevices)}"'])
for odevice in odevices:
odevice_data = odevice.split()
odevice_data_reordered = [ MAC, IP, NAME, SWITCH_AP, PORT_SSID, TYPE]
odevice_data_reordered[MAC]=odevice_data[cMAC]
odevice_data_reordered[IP]=odevice_data[cIP]
real_naxname = device_handler.getValueWithMac('dev_Name',ieee2ietf_mac_formater(odevice_data[cMAC]))
#
# if the name stored in Nax for a device is empty or the MAC addres or has some parenthhesis or is the same as in omada
# don't bother updating omada's name at all.
#
naxname = real_naxname
if real_naxname != None:
if '(' in real_naxname:
# removing parenthesis and domains from the name
naxname = real_naxname.split('(')[0]
if naxname != None and '.' in naxname:
naxname = naxname.split('.')[0]
if naxname in ( None, 'null', '' ):
naxname = odevice_data[cNAME] if odevice_data[cNAME] != '' else odevice_data[cMAC]
naxname = naxname.strip()
mylog('debug', [f'[{pluginName}] TEST name from MAC: {naxname}'])
if odevice_data[cNAME] in (odevice_data[cMAC], 'null', ''):
mylog('verbose', [f'[{pluginName}] updating omada server because odevice_data is: {odevice_data[cNAME]} and naxname is: "{naxname}"'])
callomada(['set-client-name', odevice_data[cMAC], naxname])
odevice_data_reordered[NAME] = naxname
else:
if omada_force_overwrite and naxname != odevice_data[cNAME] :
mylog('verbose', [f'[{pluginName}] updating omada server because odevice_data is: "{odevice_data[cNAME]} and naxname is: "{naxname}"'])
callomada(['set-client-name', odevice_data[cMAC], naxname])
odevice_data_reordered[NAME] = naxname
mightbeport = odevice_data[cPORT_SSID].lstrip('(')
mightbeport = mightbeport.rstrip(')')
if mightbeport.isdigit():
odevice_data_reordered[SWITCH_AP] = odevice_data[cSWITCH_AP]
odevice_data_reordered[PORT_SSID] = mightbeport
else:
odevice_data_reordered[SWITCH_AP] = mightbeport
odevice_data_reordered[PORT_SSID] = odevice_data[cSWITCH_AP]
# replacing the switch name with its MAC...
try:
mightbemac = sadevices_macbyname[odevice_data_reordered[SWITCH_AP]]
odevice_data_reordered[SWITCH_AP] = mightbemac
except KeyError:
mylog(OMDLOGLEVEL, [f'[{pluginName}] could not find the mac adddress for: "{odevice_data_reordered[SWITCH_AP]}"'])
# adding the type
odevice_data_reordered[TYPE] = 'null'
device_data_bymac[odevice_data_reordered[MAC]] = odevice_data_reordered
device_data_mac_byip[odevice_data_reordered[IP]] = odevice_data_reordered[MAC]
mylog(OMDLOGLEVEL, [f'[{pluginName}] tokens: "{odevice_data}"'])
mylog(OMDLOGLEVEL, [f'[{pluginName}] tokens_reordered: "{odevice_data_reordered}"'])
# populating the uplinks nodes of the omada switches and access points manually
# since OMADA SDN makes is unreliable if the gateway is not their own tplink hardware...
# step1 let's find the the default router
#
default_router_ip = find_default_gateway_ip()
default_router_mac = ietf2ieee_mac_formater(get_mac_from_IP(default_router_ip))
device_data_bymac[default_router_mac][TYPE] = 'Firewall'
# step2 let's find the first switch and set the default router parent to internet
first_switch=device_data_bymac[default_router_mac][SWITCH_AP]
device_data_bymac[default_router_mac][SWITCH_AP] = 'Internet'
# step3 let's set the switch connected to the default gateway uplink to the default gateway and hardcode port to 1 for now:
#device_data_bymac[first_switch][SWITCH_AP]=default_router_mac
#device_data_bymac[first_switch][SWITCH_AP][PORT_SSID] = '1'
# step4, let's go recursively through switches other links to mark update their uplinks
# and pray it ends one day...
#
add_uplink(default_router_mac,first_switch, device_data_bymac,sadevices_linksbymac,port_byswitchmac_byclientmac)
return device_data_bymac.values()
if __name__ == '__main__':
main()

195
front/plugins/omada_sdn_imp/testre.py Normal file
View File

@@ -0,0 +1,195 @@
import re
""""
how to rebuild and re-run...
savefolder=~/naxdev/NetAlertX.v7
cd ~/naxdev
mv NetAlertX $savefolder
gh repo clone FlyingToto/NetAlertX
cd NetAlertX
ln -s ../docker-compose.yml.ffsb42 .
ln -s ../.env.omada.ffsb42 .
cd front/plugins/omada_sdn_imp/
cp -p $savefoder/front/plugins/omada_sdn_imp/omada_sdn.py* .
cp -p $savefoder/front/plugins/omada_sdn_imp/README.md .
cp -p $savefoder/front/plugins/omada_sdn_imp/omada_account_sample.png .
cp -p $savefoder/front/plugins/omada_sdn_imp/testre.py .
#cp -p $savefoder/front/plugins/omada_sdn_imp/config.json config.json.v6
cd ~/naxdev/NetAlertX
sudo docker-compose --env-file .env.omada.ffsb42 -f ./docker-compose.yml.ffsb42 up
to gather data for Boris:
today=$(date +%Y_%m_%d__%H_%M)
mkdir /drives/c/temp/4boris/$today
cd /drives/c/temp/4boris/$today
scp hal:~/naxdev/logs/app.log .
scp hal:~/naxdev/NetAlertX/front/plugins/omada_sdn_imp/* .
gzip -c app.log > app_$today.log.gz
scp hal:~/naxdev/NetAlertX/front/plugins/omada_sdn_imp/omada_sdn.py /drives/c/temp/4boris/
"""
def extract_mac_addresses(text):
mac_pattern = r"([0-9A-Fa-f]{2}[:-][0-9A-Fa-f]{2}[:-][0-9A-Fa-f]{2}[:-][0-9A-Fa-f]{2}[:-][0-9A-Fa-f]{2}[:-][0-9A-Fa-f]{2})"
#mac_pattern = r'([0-9A-Fa-f]{2}[:-]){5}([0-9A-Fa-f]{2})'
#r"(([0-9A-F]{2}-){5}[0-9A-F]{2})"
#r"([0-9A-Fa-f]{2}[:-]){5}([0-9A-Fa-f]{2})"
#r"([0-9A-Fa-f]{2}[:-][0-9A-Fa-f]{2}[:-][0-9A-Fa-f]{2}[:-][0-9A-Fa-f]{2}[:-][0-9A-Fa-f]{2}[:-][0-9A-Fa-f]{2})"
mac_addresses = re.findall(mac_pattern, text)
return ["".join(parts) for parts in mac_addresses]
# Example usage:
foo = """
Name: office
Address: 0C-80-63-69-C4-D1 (192.168.0.5)
Status: CONNECTED (CONNECTED)
Ports: 28
Supports PoE: False
Model: T1600G-28TS v3.0
LED Setting: SITE_SETTINGS
Uptime: 5day(s) 22h 39m 6s
Uplink switch: D8-07-B6-71-FF-7F office24
Downlink devices:
- 40-AE-30-A5-A7-50 ompapaoffice
- B0-95-75-46-0C-39 pantry12
"""
mac_list = extract_mac_addresses(foo)
print("mac list",mac_list)
# ['0C-80-63-69-C4-D1', 'D8-07-B6-71-FF-7F', '40-AE-30-A5-A7-50', 'B0-95-75-46-0C-39']
# ['C4-:D1', 'FF-:7F', 'A7-:50', '0C-:39']
linked_switches_and_ports_by_mac = {}
foo = """"
something
some BOB12
blah BOB23
--- BEGIN ---
something else BOB12
blah BOB23
--- END ---
"""
def extract_BOB_patterns(foo):
pattern = r"BOB\d{2}(?=.*BEGIN)"
matches = re.findall(pattern, foo, re.DOTALL)
return matches
BOBresult = extract_BOB_patterns(foo)
print("BOB:",BOBresult) # Output: ['BOB12', 'BOB23']
#0C-80-63-69-C4-D1
clientmac_by_switchmac_by_switchportSSID = {}
switch_mac_and_ports_by_clientmac = {}
def extract_uplinks_mac_and_ports(tplink_device_dump):
mac_switches = []
mac_pattern = r"([0-9A-Fa-f]{2}[:-][0-9A-Fa-f]{2}[:-][0-9A-Fa-f]{2}[:-][0-9A-Fa-f]{2}[:-][0-9A-Fa-f]{2}[:-][0-9A-Fa-f]{2})(?=.*BEGIN)"
mac_addresses = re.findall(mac_pattern, tplink_device_dump,re.DOTALL)
mac_switches = ["".join(parts) for parts in mac_addresses]
print(" mac_switches1=",mac_switches)
mymac = mac_switches[0]
mylinks = mac_switches[1:]
for mylink in mylinks:
port_pattern = r"(?=\{.*\"port\"\: )([0-9]+)(?=.*"+re.escape(mylink)+r")"
port_pattern = r"(?:{/s\"port\"\: )([0-9]+)(?:[!\}].*"+re.escape(mylink)+r")"
#port_pattern = rf"{{.*?{found_mac}.*?port\s*:\s*(\d+).*?}}"
#port_pattern = rf"{{.*?.*?port\s*:\s*(\d+)[!\\}]*{mylink}?}}"
port_pattern = r"(?:\{[!\}]port/s:/s)([0-9]+\,)(?:[!\}]*"+re.escape(mylink)+r"[!\{]*\})"
#port_pattern = r"(?:\{.*\"port\"\: )([0-9]+)(?=.*"+re.escape(mylink)+r")"
port_pattern = r"(?:{[^}]*\"port\"\: )([0-9]+)(?=[^}]*"+re.escape(mylink)+r")"
myport = re.findall(port_pattern, tplink_device_dump,re.DOTALL)
print("myswitch=",mymac, "- link_switch=", mylink, "myport=", myport)
return(0)
'''
with open('/tmp/switch.bigroom.dump.json', 'r') as file:
foo3 = file_content = file.read()
print("bigroom", end="")
extract_uplinks_mac_and_ports(foo3)
with open('/tmp/switch.office.dump.json', 'r') as file:
foo4 = file_content = file.read()
print("office", end="")
extract_uplinks_mac_and_ports(foo4)
'''
import netifaces
gw = netifaces.gateways()
print(gw['default'][netifaces.AF_INET][0])
d = {'a': ['0', 'Arthur'], 'b': ['foo', 'Belling']}
print(d.items())
print(d.keys())
print(d.values())
foo = 2
#while foo > 0:
# foo = 'toto'
print("foo is ",foo)
if foo in ( 'bar', '', 'null'):
print("foo is bar")
else:
print("foo is not bar")
foo='192-168-0-150.local'
bar = foo.split('.')[0]
print("bar=",bar,"-")
bar2 = 'toto'
print("bar2=",bar2,"-")
import concurrent.futures
import time
import random
def phello(arg):
print('running phell',arg)
delay = random.uniform(0, 6)
time.sleep(delay)
return f"parallel hello : {arg}", delay
def testparalel():
arguments = ["Alice", "Bob", "Charlie", "David"]
results = {}
results2 = {}
para = 10
# Using ThreadPoolExecutor for parallel execution
with concurrent.futures.ThreadPoolExecutor(max_workers=para) as executor:
# Submit tasks to the executor
future_to_arg = {executor.submit(phello, arg): arg for arg in arguments}
# Wait for all futures to complete
done, _ = concurrent.futures.wait(future_to_arg)
# Retrieve results
for future in done:
arg = future_to_arg[future]
try:
result, result2 = future.result()
results[arg] = result
results2[arg] = result2
except Exception as exc:
print(f"{arg} generated an exception: {exc}")
# Print results after all threads have completed
print("All threads completed. Results:")
for arg, result in results.items():
print(f"arg:{arg}, result={results[arg]}, result2={results2[arg]}")
testparalel()