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

Name matching fixes 🩹

Browse Source
This commit is contained in:
Jokob-sk
2023-11-04 12:06:12 +11:00
parent c8e494596e
commit 973cd60893
7 changed files with 190 additions and 102 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
dockerfiles/start.sh
View File

@@ -86,7 +86,6 @@ fi
echo "[INSTALL] Copy starter pialert.db and pialert.conf if they don't exist"
# Copy starter pialert.db and pialert.conf if they don't exist
# cp -n "/home/pi/pialert/back/pialert.conf" "/home/pi/pialert/config/pialert.conf"
cp -n "$INSTALL_DIR/pialert/back/pialert.conf" "$INSTALL_DIR/pialert/config/pialert.conf"
cp -n "$INSTALL_DIR/pialert/back/pialert.db" "$INSTALL_DIR/pialert/db/pialert.db"

21
front/plugins/ddns_update/script.py
View File

@@ -28,9 +28,11 @@ 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')
pluginName = 'DDNS'
def main():
mylog('verbose', ['[DDNS] In script'])
mylog('verbose', [f'[{pluginName}] In script'])
parser = argparse.ArgumentParser(description='Check internet connectivity and IP')
@@ -52,7 +54,7 @@ def main():
# perform the new IP lookup and DDNS tasks if enabled
ddns_update( DDNS_UPDATE_URL, DDNS_USER, DDNS_PASSWORD, DDNS_DOMAIN, PREV_IP)
mylog('verbose', ['[DDNS] Finished '])
mylog('verbose', [f'[{pluginName}] Finished '])
return 0
@@ -65,20 +67,20 @@ def ddns_update ( DDNS_UPDATE_URL, DDNS_USER, DDNS_PASSWORD, DDNS_DOMAIN, PREV_I
# Update DDNS record if enabled and IP is different
# Get Dynamic DNS IP
mylog('verbose', ['[DDNS] Retrieving Dynamic DNS IP'])
mylog('verbose', [f'[{pluginName}] Retrieving Dynamic DNS IP'])
dns_IP = get_dynamic_DNS_IP(DDNS_DOMAIN)
# Check Dynamic DNS IP
if dns_IP == "" or dns_IP == "0.0.0.0" :
mylog('none', ['[DDNS] Error retrieving Dynamic DNS IP'])
mylog('none', [f'[{pluginName}] Error retrieving Dynamic DNS IP'])
mylog('none', ['[DDNS] ', dns_IP])
mylog('none', [f'[{pluginName}] ', dns_IP])
# Check DNS Change
if dns_IP != PREV_IP :
mylog('none', ['[DDNS] Updating Dynamic DNS IP'])
mylog('none', [f'[{pluginName}] Updating Dynamic DNS IP'])
message = set_dynamic_DNS_IP (DDNS_UPDATE_URL, DDNS_USER, DDNS_PASSWORD, DDNS_DOMAIN)
mylog('none', ['[DDNS] ', message])
mylog('none', [f'[{pluginName}] ', message])
# plugin_objects = Plugin_Objects(RESULT_FILE)
@@ -104,9 +106,10 @@ def get_dynamic_DNS_IP (DDNS_DOMAIN):
try:
# try runnning a subprocess
dig_output = subprocess.check_output (dig_args, universal_newlines=True)
mylog('none', [f'[{pluginName}] DIG output :', dig_output])
except subprocess.CalledProcessError as e:
# An error occured, handle it
mylog('none', ['[DDNS] ERROR - ', e.output])
mylog('none', [f'[{pluginName}] ERROR - ', e.output])
dig_output = '' # probably no internet
# Check result is an IP
@@ -132,7 +135,7 @@ def set_dynamic_DNS_IP (DDNS_UPDATE_URL, DDNS_USER, DDNS_PASSWORD, DDNS_DOMAIN):
universal_newlines=True)
except subprocess.CalledProcessError as e:
# An error occured, handle it
mylog('none', ['[DDNS] ERROR - ',e.output])
mylog('none', [f'[{pluginName}] ERROR - ',e.output])
curl_output = ""
return curl_output

17
front/plugins/internet_ip/config.json
View File

@@ -46,7 +46,7 @@
"string": "A plugin to check your internet connectivity and IP."
},
{
"language_code": "en_us",
"language_code": "de_de",
"string": "Ein Plugin zur Prüfung der Internetverbindung und externen IP."
}
],
@@ -382,6 +382,21 @@
}
]
},
{
"column": "Watched_Value2",
"css_classes": "col-sm-2",
"show": true,
"type": "textarea_readonly",
"default_value":"",
"options": [],
"localized": ["name"],
"name":[
{
"language_code": "en_us",
"string" : "Response"
}
]
},
{
"column": "Dummy",
"mapped_to_column": "cur_ScanMethod",

31
front/plugins/internet_ip/script.py
View File

@@ -28,9 +28,11 @@ 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')
pluginName = 'INTRNT'
def main():
mylog('verbose', ['[INTRNT] In script'])
mylog('verbose', [f'[{pluginName}] In script'])
parser = argparse.ArgumentParser(description='Check internet connectivity and IP')
@@ -42,15 +44,15 @@ def main():
PREV_IP = values.prev_ip.split('=')[1]
DIG_GET_IP_ARG = values.DIG_GET_IP_ARG.split('=b')[1] # byte64 encoded
mylog('verbose', ['[INTRNT] DIG_GET_IP_ARG: ', DIG_GET_IP_ARG])
mylog('verbose', [f'[{pluginName}] DIG_GET_IP_ARG: ', DIG_GET_IP_ARG])
# Decode the base64-encoded value to get the actual value in ASCII format.
DIG_GET_IP_ARG = base64.b64decode(DIG_GET_IP_ARG).decode('ascii')
mylog('verbose', [f'[INTRNT] DIG_GET_IP_ARG resolved: {DIG_GET_IP_ARG} '])
mylog('verbose', [f'[{pluginName}] DIG_GET_IP_ARG resolved: {DIG_GET_IP_ARG} '])
# perform the new IP lookup
new_internet_IP = check_internet_IP( PREV_IP, DIG_GET_IP_ARG)
new_internet_IP, cmd_output = check_internet_IP( PREV_IP, DIG_GET_IP_ARG)
plugin_objects = Plugin_Objects(RESULT_FILE)
@@ -58,7 +60,7 @@ def main():
primaryId = 'Internet', # MAC (Device Name)
secondaryId = new_internet_IP, # IP Address
watched1 = f'Previous IP: {PREV_IP}',
watched2 = '',
watched2 = cmd_output.replace('\n',''),
watched3 = '',
watched4 = '',
extra = f'Previous IP: {PREV_IP}',
@@ -66,7 +68,7 @@ def main():
plugin_objects.write_result_file()
mylog('verbose', ['[INTRNT] Finished '])
mylog('verbose', [f'[{pluginName}] Finished '])
return 0
@@ -77,10 +79,10 @@ def main():
def check_internet_IP ( PREV_IP, DIG_GET_IP_ARG ):
# Get Internet IP
mylog('verbose', ['[INTRNT] - Retrieving Internet IP'])
internet_IP = get_internet_IP(DIG_GET_IP_ARG)
mylog('verbose', [f'[{pluginName}] - Retrieving Internet IP'])
internet_IP, cmd_output = get_internet_IP(DIG_GET_IP_ARG)
mylog('verbose', [f'[INTRNT] Current internet_IP : {internet_IP}'])
mylog('verbose', [f'[{pluginName}] Current internet_IP : {internet_IP}'])
# Check previously stored IP
previous_IP = '0.0.0.0'
@@ -88,23 +90,26 @@ def check_internet_IP ( PREV_IP, DIG_GET_IP_ARG ):
if PREV_IP is not None and len(PREV_IP) > 0 :
previous_IP = PREV_IP
mylog('verbose', [f'[INTRNT] previous_IP : {previous_IP}'])
mylog('verbose', [f'[{pluginName}] previous_IP : {previous_IP}'])
# logging
append_line_to_file (logPath + '/IP_changes.log', '['+str(timeNowTZ()) +']\t'+ internet_IP +'\n')
return internet_IP
return internet_IP, cmd_output
#-------------------------------------------------------------------------------
def get_internet_IP (DIG_GET_IP_ARG):
cmd_output = ''
# Using 'dig'
dig_args = ['dig', '+short'] + DIG_GET_IP_ARG.strip().split()
try:
cmd_output = subprocess.check_output (dig_args, universal_newlines=True)
mylog('verbose', [f'[{pluginName}] DIG result : {cmd_output}'])
except subprocess.CalledProcessError as e:
mylog('none', [e.output])
mylog('verbose', [e.output])
cmd_output = '' # no internet
# Check result is an IP
@@ -114,7 +119,7 @@ def get_internet_IP (DIG_GET_IP_ARG):
if IP == '':
IP = '0.0.0.0'
return IP
return IP, cmd_output
#===============================================================================
# BEGIN

2
front/plugins/undiscoverables/config.json
View File

@@ -49,7 +49,7 @@
"string": "Este complemento es para importar dispositivos no detectables desde un archivo."
},
{
"language_code": "en_us",
"language_code": "de_de",
"string": "Ein Plugin zum Importieren von nicht erkennbaren Geräten aus einer Datei."
}
],

20
pialert/device.py
View File

@@ -247,6 +247,8 @@ def update_devices_names (db):
recordsToUpdate = []
recordsNotFound = []
nameNotFound = "(name not found)"
ignored = 0
notFound = 0
@@ -274,27 +276,31 @@ def update_devices_names (db):
mylog('verbose', ['[Update Device Name] Pholus entries from prev scans: ', len(pholusResults)])
for device in unknownDevices:
newName = -1
newName = nameNotFound
# Resolve device name with DiG
newName = resolve_device_name_dig (device['dev_MAC'], device['dev_LastIP'])
# count
if newName != -1:
if newName != nameNotFound:
foundDig += 1
# Resolve with Pholus
if newName == -1:
newName = resolve_device_name_pholus (device['dev_MAC'], device['dev_LastIP'], pholusResults)
if newName == nameNotFound:
newName = resolve_device_name_pholus (device['dev_MAC'], device['dev_LastIP'], pholusResults, nameNotFound, False)
# Try IP matching only
if newName == nameNotFound:
newName = resolve_device_name_pholus (device['dev_MAC'], device['dev_LastIP'], pholusResults, nameNotFound, True)
# count
if newName != -1:
if newName != nameNotFound:
foundPholus += 1
# isf still not found update name so we can distinguish the devices where we tried already
if newName == -1 :
if newName == nameNotFound :
# if dev_Name is the same as what we will change it to, take no action
# this mitigates a race condition which would overwrite a users edits that occured since the select earlier
if device['dev_Name'] != "(name not found)":
if device['dev_Name'] != nameNotFound:
recordsNotFound.append (["(name not found)", device['dev_MAC']])
else:
# name was found with DiG or Pholus

148
pialert/helper.py
View File

@@ -329,12 +329,16 @@ def checkIPV4(ip):
#-------------------------------------------------------------------------------
def check_IP_format (pIP):
# check if TCP communication error ocurred
if 'communications error to' in pIP:
return ''
# Check IP format
IPv4SEG = r'(?:25[0-5]|(?:2[0-4]|1{0,1}[0-9]){0,1}[0-9])'
IPv4ADDR = r'(?:(?:' + IPv4SEG + r'\.){3,3}' + IPv4SEG + r')'
IP = re.search(IPv4ADDR, pIP)
# Return error if not IP
# Return empty if not IP
if IP is None :
return ""
@@ -346,39 +350,37 @@ def check_IP_format (pIP):
#-------------------------------------------------------------------------------
def resolve_device_name_dig (pMAC, pIP):
newName = ""
nameNotFound = "(name not found)"
try :
dig_args = ['dig', '+short', '-x', pIP]
# Execute command
try:
# try runnning a subprocess
newName = subprocess.check_output (dig_args, universal_newlines=True)
except subprocess.CalledProcessError as e:
# An error occured, handle it
mylog('none', ['[device_name_dig] ', e.output])
# newName = "Error - check logs"
return -1
# Check returns
newName = newName.strip()
if len(newName) == 0 :
return -1
return nameNotFound
# Cleanup
newName = cleanResult(newName)
newName = cleanDeviceName(newName, True)
if newName == "" or len(newName) == 0:
return -1
if newName == "" or len(newName) == 0 or newName == '-1' or newName == -1 or "communications error" in newName:
return nameNotFound
# all checks passed
mylog('debug', [f'[resolve_device_name_dig] Found a new name: "{newName}"'])
# Return newName
return newName
# not Found
except subprocess.CalledProcessError :
return -1
except subprocess.CalledProcessError as e:
# An error occured, handle it
mylog('none', ['[resolve_device_name_dig] ERROR: ', e.output])
# newName = "Error - check logs"
return nameNotFound
#-------------------------------------------------------------------------------
@@ -389,14 +391,17 @@ def resolve_device_name_dig (pMAC, pIP):
# Disclaimer - I'm interfacing with a script I didn't write (pholus3.py) so it's possible I'm missing types of answers
# it's also possible the pholus3.py script can be adjusted to provide a better output to interface with it
# Hit me with a PR if you know how! :)
def resolve_device_name_pholus (pMAC, pIP, allRes):
def resolve_device_name_pholus (pMAC, pIP, allRes, nameNotFound, matchIpOnly = False):
pholusMatchesIndexes = []
result = nameNotFound
# Collect all Pholus entries with matching MAC and of type Answer
index = 0
for result in allRes:
# limiting entries used for name resolution to the ones containing the current IP (v4 only)
if result["MAC"] == pMAC and result["Record_Type"] == "Answer" and result["IP_v4_or_v6"] == pIP and '._googlezone' not in result["Value"]:
if ((matchIpOnly and result["IP_v4_or_v6"] == pIP ) or ( result["MAC"] == pMAC )) and result["Record_Type"] == "Answer" and result["IP_v4_or_v6"] == pIP and '._googlezone' not in result["Value"]:
# found entries with a matching MAC address, let's collect indexes
pholusMatchesIndexes.append(index)
@@ -404,66 +409,117 @@ def resolve_device_name_pholus (pMAC, pIP, allRes):
# return if nothing found
if len(pholusMatchesIndexes) == 0:
return -1
return nameNotFound
# we have some entries let's try to select the most useful one
# Do I need to pre-order allRes to have the most valuable onse on the top?
for i in pholusMatchesIndexes:
if not checkIPV4(allRes[i]['IP_v4_or_v6']):
continue
value = allRes[i]["Value"]
# airplay matches contain a lot of information
# Matches for example:
# Brand Tv (50)._airplay._tcp.local. TXT Class:32769 "acl=0 deviceid=66:66:66:66:66:66 features=0x77777,0x38BCB46 rsf=0x3 fv=p20.T-FFFFFF-03.1 flags=0x204 model=XXXX manufacturer=Brand serialNumber=XXXXXXXXXXX protovers=1.1 srcvers=777.77.77 pi=FF:FF:FF:FF:FF:FF psi=00000000-0000-0000-0000-FFFFFFFFFF gid=00000000-0000-0000-0000-FFFFFFFFFF gcgl=0 pk=AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA"
for i in pholusMatchesIndexes:
if checkIPV4(allRes[i]['IP_v4_or_v6']) and '._airplay._tcp.local. TXT Class:32769' in str(allRes[i]["Value"]) :
return allRes[i]["Value"].split('._airplay._tcp.local. TXT Class:32769')[0]
if '._airplay._tcp.local. TXT Class:32769' in value:
return cleanDeviceName(value.split('._airplay._tcp.local. TXT Class:32769')[0], matchIpOnly)
# second best - contains airplay
# Matches for example:
# _airplay._tcp.local. PTR Class:IN "Brand Tv (50)._airplay._tcp.local."
for i in pholusMatchesIndexes:
if checkIPV4(allRes[i]['IP_v4_or_v6']) and '_airplay._tcp.local. PTR Class:IN' in allRes[i]["Value"] and ('._googlecast') not in allRes[i]["Value"]:
return cleanResult(allRes[i]["Value"].split('"')[1])
if '_airplay._tcp.local. PTR Class:IN' in value and ('._googlecast') not in value:
return cleanDeviceName(value.split('"')[1], matchIpOnly)
# Contains PTR Class:32769
# Matches for example:
# 3.1.168.192.in-addr.arpa. PTR Class:32769 "MyPc.local."
for i in pholusMatchesIndexes:
if checkIPV4(allRes[i]['IP_v4_or_v6']) and 'PTR Class:32769' in allRes[i]["Value"]:
return cleanResult(allRes[i]["Value"].split('"')[1])
if 'PTR Class:32769' in value:
return cleanDeviceName(value.split('"')[1], matchIpOnly)
# Contains AAAA Class:IN
# Matches for example:
# DESKTOP-SOMEID.local. AAAA Class:IN "fe80::fe80:fe80:fe80:fe80"
for i in pholusMatchesIndexes:
if checkIPV4(allRes[i]['IP_v4_or_v6']) and 'AAAA Class:IN' in allRes[i]["Value"]:
return cleanResult(allRes[i]["Value"].split('.local.')[0])
if 'AAAA Class:IN' in value:
return cleanDeviceName(value.split('.local.')[0], matchIpOnly)
# Contains _googlecast._tcp.local. PTR Class:IN
# Matches for example:
# _googlecast._tcp.local. PTR Class:IN "Nest-Audio-ff77ff77ff77ff77ff77ff77ff77ff77._googlecast._tcp.local."
for i in pholusMatchesIndexes:
if checkIPV4(allRes[i]['IP_v4_or_v6']) and '_googlecast._tcp.local. PTR Class:IN' in allRes[i]["Value"] and ('Google-Cast-Group') not in allRes[i]["Value"]:
return cleanResult(allRes[i]["Value"].split('"')[1])
if '_googlecast._tcp.local. PTR Class:IN' in value and ('Google-Cast-Group') not in value:
return cleanDeviceName(value.split('"')[1], matchIpOnly)
# Contains A Class:32769
# Matches for example:
# Android.local. A Class:32769 "192.168.1.6"
for i in pholusMatchesIndexes:
if checkIPV4(allRes[i]['IP_v4_or_v6']) and ' A Class:32769' in allRes[i]["Value"]:
return cleanResult(allRes[i]["Value"].split(' A Class:32769')[0])
if ' A Class:32769' in value:
return cleanDeviceName(value.split(' A Class:32769')[0], matchIpOnly)
# # Contains PTR Class:IN
# Contains PTR Class:IN
# Matches for example:
# _esphomelib._tcp.local. PTR Class:IN "ceiling-light-1._esphomelib._tcp.local."
for i in pholusMatchesIndexes:
if checkIPV4(allRes[i]['IP_v4_or_v6']) and 'PTR Class:IN' in allRes[i]["Value"]:
if allRes[i]["Value"] and len(allRes[i]["Value"].split('"')) > 1:
return cleanResult(allRes[i]["Value"].split('"')[1])
if 'PTR Class:IN' in value and len(value.split('"')) > 1:
return cleanDeviceName(value.split('"')[1], matchIpOnly)
return -1
# # airplay matches contain a lot of information
# # Matches for example:
# # Brand Tv (50)._airplay._tcp.local. TXT Class:32769 "acl=0 deviceid=66:66:66:66:66:66 features=0x77777,0x38BCB46 rsf=0x3 fv=p20.T-FFFFFF-03.1 flags=0x204 model=XXXX manufacturer=Brand serialNumber=XXXXXXXXXXX protovers=1.1 srcvers=777.77.77 pi=FF:FF:FF:FF:FF:FF psi=00000000-0000-0000-0000-FFFFFFFFFF gid=00000000-0000-0000-0000-FFFFFFFFFF gcgl=0 pk=AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA"
# for i in pholusMatchesIndexes:
# if checkIPV4(allRes[i]['IP_v4_or_v6']) and '._airplay._tcp.local. TXT Class:32769' in str(allRes[i]["Value"]) :
# return cleanDeviceName(allRes[i]["Value"].split('._airplay._tcp.local. TXT Class:32769')[0], matchIpOnly)
# # second best - contains airplay
# # Matches for example:
# # _airplay._tcp.local. PTR Class:IN "Brand Tv (50)._airplay._tcp.local."
# for i in pholusMatchesIndexes:
# if checkIPV4(allRes[i]['IP_v4_or_v6']) and '_airplay._tcp.local. PTR Class:IN' in allRes[i]["Value"] and ('._googlecast') not in allRes[i]["Value"]:
# return cleanDeviceName(allRes[i]["Value"].split('"')[1], matchIpOnly)
# # Contains PTR Class:32769
# # Matches for example:
# # 3.1.168.192.in-addr.arpa. PTR Class:32769 "MyPc.local."
# for i in pholusMatchesIndexes:
# if checkIPV4(allRes[i]['IP_v4_or_v6']) and 'PTR Class:32769' in allRes[i]["Value"]:
# return cleanDeviceName(allRes[i]["Value"].split('"')[1], matchIpOnly)
# # Contains AAAA Class:IN
# # Matches for example:
# # DESKTOP-SOMEID.local. AAAA Class:IN "fe80::fe80:fe80:fe80:fe80"
# for i in pholusMatchesIndexes:
# if checkIPV4(allRes[i]['IP_v4_or_v6']) and 'AAAA Class:IN' in allRes[i]["Value"]:
# return cleanDeviceName(allRes[i]["Value"].split('.local.')[0], matchIpOnly)
# # Contains _googlecast._tcp.local. PTR Class:IN
# # Matches for example:
# # _googlecast._tcp.local. PTR Class:IN "Nest-Audio-ff77ff77ff77ff77ff77ff77ff77ff77._googlecast._tcp.local."
# for i in pholusMatchesIndexes:
# if checkIPV4(allRes[i]['IP_v4_or_v6']) and '_googlecast._tcp.local. PTR Class:IN' in allRes[i]["Value"] and ('Google-Cast-Group') not in allRes[i]["Value"]:
# return cleanDeviceName(allRes[i]["Value"].split('"')[1], matchIpOnly)
# # Contains A Class:32769
# # Matches for example:
# # Android.local. A Class:32769 "192.168.1.6"
# for i in pholusMatchesIndexes:
# if checkIPV4(allRes[i]['IP_v4_or_v6']) and ' A Class:32769' in allRes[i]["Value"]:
# return cleanDeviceName(allRes[i]["Value"].split(' A Class:32769')[0], matchIpOnly)
# # # Contains PTR Class:IN
# # Matches for example:
# # _esphomelib._tcp.local. PTR Class:IN "ceiling-light-1._esphomelib._tcp.local."
# for i in pholusMatchesIndexes:
# if checkIPV4(allRes[i]['IP_v4_or_v6']) and 'PTR Class:IN' in allRes[i]["Value"]:
# if allRes[i]["Value"] and len(allRes[i]["Value"].split('"')) > 1:
# return cleanDeviceName(allRes[i]["Value"].split('"')[1], matchIpOnly)
return nameNotFound
#-------------------------------------------------------------------------------
def cleanResult(str):
def cleanDeviceName(str, matchIpOnly):
# alternative str.split('.')[0]
str = str.replace("._airplay", "")
str = str.replace("._tcp", "")
@@ -478,6 +534,10 @@ def cleanResult(str):
if str.endswith('.'):
str = str[:-1]
if matchIpOnly:
str = str + " (IP match)"
return str