jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-07-19 03:07:37 +00:00
Code Issues Packages Projects Releases Wiki Activity

Restored telnet as an option. Activate with config option 'protocol: telnet'. Default is ssh (#2096)

Browse Source
This commit is contained in:
froz 2016-05-17 15:55:12 -07:00 committed by Paulus Schoutsen
parent 15f89fc636
commit 8d34b76d51
1 changed files with 40 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

45
homeassistant/components/device_tracker/asuswrt.py
View File

@ -6,6 +6,7 @@ https://home-assistant.io/components/device_tracker.asuswrt/
"""
import logging
import re
import telnetlib
import threading
from datetime import timedelta
@ -20,12 +21,14 @@ MIN_TIME_BETWEEN_SCANS = timedelta(seconds=5)
_LOGGER = logging.getLogger(__name__)
REQUIREMENTS = ['pexpect==4.0.1']
_LEASES_CMD = 'cat /var/lib/misc/dnsmasq.leases'
_LEASES_REGEX = re.compile(
r'\w+\s' +
r'(?P<mac>(([0-9a-f]{2}[:-]){5}([0-9a-f]{2})))\s' +
r'(?P<ip>([0-9]{1,3}[\.]){3}[0-9]{1,3})\s' +
r'(?P<host>([^\s]+))')
_IP_NEIGH_CMD = 'ip neigh'
_IP_NEIGH_REGEX = re.compile(
r'(?P<ip>([0-9]{1,3}[\.]){3}[0-9]{1,3})\s' +
r'\w+\s' +
@ -55,6 +58,7 @@ class AsusWrtDeviceScanner(object):
self.host = config[CONF_HOST]
self.username = str(config[CONF_USERNAME])
self.password = str(config[CONF_PASSWORD])
self.protocol = config.get('protocol')
self.lock = threading.Lock()
@ -100,22 +104,53 @@ class AsusWrtDeviceScanner(object):
self.last_results = active_clients
return True
def get_asuswrt_data(self):
"""Retrieve data from ASUSWRT and return parsed result."""
def ssh_connection(self):
"""Retrieve data from ASUSWRT via the ssh protocol."""
from pexpect import pxssh
try:
ssh = pxssh.pxssh()
ssh.login(self.host, self.username, self.password)
ssh.sendline('ip neigh')
ssh.sendline(_IP_NEIGH_CMD)
ssh.prompt()
neighbors = ssh.before.split(b'\n')[1:-1]
ssh.sendline('cat /var/lib/misc/dnsmasq.leases')
ssh.sendline(_LEASES_CMD)
ssh.prompt()
leases_result = ssh.before.split(b'\n')[1:-1]
ssh.logout()
return (neighbors, leases_result)
except pxssh.ExceptionPxssh as exc:
_LOGGER.exception('Unexpected response from router: %s', exc)
return
return ('', '')
def telnet_connection(self):
"""Retrieve data from ASUSWRT via the telnet protocol."""
try:
telnet = telnetlib.Telnet(self.host)
telnet.read_until(b'login: ')
telnet.write((self.username + '\n').encode('ascii'))
telnet.read_until(b'Password: ')
telnet.write((self.password + '\n').encode('ascii'))
prompt_string = telnet.read_until(b'#').split(b'\n')[-1]
telnet.write('{}\n'.format(_IP_NEIGH_CMD).encode('ascii'))
neighbors = telnet.read_until(prompt_string).split(b'\n')[1:-1]
telnet.write('{}\n'.format(_LEASES_CMD).encode('ascii'))
leases_result = telnet.read_until(prompt_string).split(b'\n')[1:-1]
telnet.write('exit\n'.encode('ascii'))
return (neighbors, leases_result)
except EOFError:
_LOGGER.exception("Unexpected response from router")
return ('', '')
except ConnectionRefusedError:
_LOGGER.exception("Connection refused by router,"
" is telnet enabled?")
return ('', '')
def get_asuswrt_data(self):
"""Retrieve data from ASUSWRT and return parsed result."""
if self.protocol == 'telnet':
neighbors, leases_result = self.telnet_connection()
else:
neighbors, leases_result = self.ssh_connection()
devices = {}
for lease in leases_result: