From d8031f220346a0786107f5e5524abd7fc9c3f558 Mon Sep 17 00:00:00 2001 From: thevickypedia Date: Wed, 29 Nov 2023 20:18:18 -0600 Subject: [PATCH] Add support for Netgear routers Setup customized logger Update requirements.txt and README.md --- README.md | 2 +- netfuse/__init__.py | 2 +- netfuse/config.py | 6 ++++++ netfuse/logger.py | 5 ++++- netfuse/main.py | 27 +++++++++++++++------------ netfuse/modules/att.py | 2 +- netfuse/modules/netgear.py | 25 +++++++++++++++++++++++++ netfuse/requirements.txt | 16 ++++++++++++---- 8 files changed, 65 insertions(+), 20 deletions(-) create mode 100644 netfuse/modules/netgear.py diff --git a/README.md b/README.md index b6ace8e..d61da41 100644 --- a/README.md +++ b/README.md @@ -2,7 +2,7 @@ NetFuse is a python module to dump hostname and IP address mapping for localhost into the hosts file -> :warning:   Currently works only for At&t users +> :warning:   To use this module, the router should be `Netgear`, [OR] the ISP should be `At&t` ### Installation ```shell diff --git a/netfuse/__init__.py b/netfuse/__init__.py index a272106..5edf828 100644 --- a/netfuse/__init__.py +++ b/netfuse/__init__.py @@ -1,3 +1,3 @@ """Place holder package""" -version = "0.0.1" +version = "0.0.2" diff --git a/netfuse/config.py b/netfuse/config.py index 1333cf5..f691501 100644 --- a/netfuse/config.py +++ b/netfuse/config.py @@ -1,8 +1,13 @@ import platform import time +from os import environ from typing import Callable, Tuple +class ValidationError(ValueError): + """Custom validation error.""" + + class Settings: """Wrapper for the settings required by the module. @@ -23,6 +28,7 @@ class Settings: flush_dns: Tuple = tuple() # Unlike Windows and macOS, Ubuntu and Linux Mint do not cache DNS queries at the operating system level by default. start = time.time() + router_pass = environ.get('router_pass') or environ.get('ROUTER_PASS') settings = Settings() diff --git a/netfuse/logger.py b/netfuse/logger.py index c0b5e0f..a8106ad 100644 --- a/netfuse/logger.py +++ b/netfuse/logger.py @@ -1,5 +1,8 @@ import logging LOGGER = logging.getLogger(__name__) -LOGGER.addHandler(logging.StreamHandler()) +_HANDLER = logging.StreamHandler() +_FORMATTER = logging.Formatter('%(asctime)s - %(levelname)s - [%(module)s:%(lineno)d] - %(funcName)s - %(message)s') +_HANDLER.setFormatter(_FORMATTER) +LOGGER.addHandler(_HANDLER) LOGGER.setLevel(logging.INFO) diff --git a/netfuse/main.py b/netfuse/main.py index 456a920..6dcda35 100644 --- a/netfuse/main.py +++ b/netfuse/main.py @@ -4,10 +4,9 @@ import click -from netfuse import version as pkg_version -from netfuse.config import settings +from netfuse.config import settings, ValidationError from netfuse.logger import LOGGER -from netfuse.modules import att +from netfuse.modules import att, netgear def parse_host_file(filepath) -> Dict[Union[int, str], Union[List[str], str]]: @@ -92,14 +91,14 @@ def flush_dns_cache() -> None: LOGGER.info("Finished updating hosts file in %.2fs", time.time() - settings.start) -def dump(dry_run: bool, filepath: str, output: str) -> None: +def dump(dry_run: bool, filepath: str, output: str, module: Union[att, netgear]) -> None: """Dumps all devices' hostname and IP addresses into the hosts file.""" host_entries = parse_host_file(filepath) - for device in att.get_attached_devices(): + for device in module.attached_devices(): if device.ipv4_address: host_entries[device.ipv4_address] = device.name else: - LOGGER.error(device.__dict__) + LOGGER.warning("%s [%s] does not have an IP address", device.name, device.mac_address) update_host_file(host_entries, output, dry_run) if output == settings.etc_hosts: flush_dns_cache() @@ -107,17 +106,21 @@ def dump(dry_run: bool, filepath: str, output: str) -> None: @click.command() @click.pass_context -@click.option("-v", "--version", required=False, is_flag=True, help="Get version of the package") +@click.option("-m", "--model", required=False, help="Source model that's either 'att' or 'netgear'") @click.option("-d", "--dry", required=False, is_flag=True, help="Dry run without updating the hosts file") @click.option("-p", "--path", required=False, default=settings.etc_hosts, help=f"Path for the hosts file, defaults to: {settings.etc_hosts}") @click.option("-o", "--output", required=False, default=settings.etc_hosts, help=f"Output filepath to write the updated entries, defaults to: {settings.etc_hosts}") -def main(*args, version: bool = False, dry: bool = False, path: str = None, output: str = None): - if version: - print(pkg_version) - return - dump(dry_run=dry, filepath=path, output=output) +def main(*args, model: str, dry: bool = False, path: str = None, output: str = None): + mapped = {"att": att, "netgear": netgear} + if model in mapped.keys(): + dump(dry_run=dry, filepath=path, output=output, module=mapped[model]) + else: + raise ValidationError( + "\n\nmodel\n Input should be 'att' or 'netgear' " + f"[type=string_type, input_value={model}, input_type={type(model)}]\n" + ) if __name__ == '__main__': diff --git a/netfuse/modules/att.py b/netfuse/modules/att.py index 1e35a92..23efd34 100644 --- a/netfuse/modules/att.py +++ b/netfuse/modules/att.py @@ -46,7 +46,7 @@ def generate_dataframe() -> pd.DataFrame: LOGGER.error("[%s] - %s" % (response.status_code, response.text)) -def get_attached_devices() -> Generator[Device]: +def attached_devices() -> Generator[Device]: """Get all devices connected to the router. Yields: diff --git a/netfuse/modules/netgear.py b/netfuse/modules/netgear.py new file mode 100644 index 0000000..70940bc --- /dev/null +++ b/netfuse/modules/netgear.py @@ -0,0 +1,25 @@ +from typing import List + +from pynetgear import Device, Netgear + +from netfuse.config import settings, ValidationError +from netfuse.logger import LOGGER + + +def attached_devices() -> List[Device]: + """Get all attached devices in a Netgear router. + + Returns: + List[Device]: + List of device objects. + """ + if not settings.router_pass: + raise ValidationError( + "\n\nrouter_pass\n Input should be a valid string " + f"[type=string_type, input_value={settings.router_pass}, input_type={type(settings.router_pass)}]\n" + ) + netgear = Netgear(password=settings.router_pass) + if devices := netgear.get_attached_devices(): + return devices + else: + LOGGER.error("Unable to get attached devices.") diff --git a/netfuse/requirements.txt b/netfuse/requirements.txt index b045e99..9a21ee4 100644 --- a/netfuse/requirements.txt +++ b/netfuse/requirements.txt @@ -1,4 +1,12 @@ -pandas -requests -lxml -click +# required by default +click==8.1.7 + +# FIX ME: Requirements are too generic, make it specific with installation netfuse[att] or netfuse[netgear] + +# required for At&t +pandas==2.1.3 +lxml==4.9.3 + +# required for Netgear users +requests>=2.31.0 +pynetgear==0.10.10