diff --git a/__init__.py b/__init__.py deleted file mode 100644 index e69de29..0000000 diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..9c8c778 --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,4 @@ +[build-system] +build-backend = "setuptools.build_meta" +requires = ["setuptools", "wheel"] + diff --git a/setup.cfg b/setup.cfg new file mode 100644 index 0000000..1fe65cf --- /dev/null +++ b/setup.cfg @@ -0,0 +1,17 @@ +[metadata] +name = xmo-remote-client +version = 0.0.1 + +[options] +packages = + xmo +install_requires = + anyio + asyncclick + sagemcom_api + pyaml + +[options.entry_points] +console_scripts = + xmo-remote-client = xmo.__main__:main + diff --git a/setup.py b/setup.py new file mode 100644 index 0000000..ecd37c5 --- /dev/null +++ b/setup.py @@ -0,0 +1,4 @@ +from setuptools import setup + +setup() + diff --git a/xmo-remote-client.py b/xmo-remote-client.py deleted file mode 100755 index 2f5a184..0000000 --- a/xmo-remote-client.py +++ /dev/null @@ -1,171 +0,0 @@ -from aiohttp import ClientSession, ClientTimeout -from aiohttp.connector import TCPConnector -import asyncclick as click -from enum import Enum -from ipaddress import IPv4Address -import json -import os -import re -from sagemcom_api.client import SagemcomClient -from sagemcom_api.enums import EncryptionMethod -import yaml - - -class EnumChoice(click.Choice): - def __init__(self, enum: Enum, case_sensitive=False): - self.__enum = enum - super().__init__(choices=[item.value for item in enum], case_sensitive=case_sensitive) - - def convert(self, value, param, ctx): - if value is None or isinstance(value, Enum): - return value - converted_str = super().convert(value, param, ctx) - return self.__enum(converted_str) - - -@click.group(chain=True) -@click.option('-H', '--host', default='192.168.2.1', help='Hostname or host IP', type=IPv4Address) -@click.option('-u', '--username', default='admin', help='Administrator username') -@click.option('-p', '--password', required=True, help='Administrator password', prompt=True, hide_input=True) -@click.option('-a', '--authentication-method', - default=EncryptionMethod.SHA512, type=EnumChoice(EncryptionMethod), - help='Authentication method') -@click.pass_context -async def cli(ctx: click.Context, host: str, username: str, password: str, authentication_method: EncryptionMethod) -> None: - ctx.obj = client = await ctx.with_async_resource( - SagemcomClient(host, username, password, authentication_method, - ClientSession( - headers={"User-Agent": "XMO_REMOTE_CLIENT/1.0.0"}, - timeout=ClientTimeout(), - connector=TCPConnector(ssl=True), - ), True, keep_keys=True - ) - ) - try: - await client.login() - except Exception as e: - click.echo(e, err=True) - raise click.Abort() - - -@cli.command() -@click.option('--path', required=True, multiple=True) -@click.pass_context -async def get_value(ctx: click.Context, path: str) -> None: - client = ctx.find_object(SagemcomClient) - if client is None: - return - for _path in path: - try: - value = await client.get_value_by_xpath(_path) - except Exception as e: - click.echo(e, err=True) - raise click.Abort() - else: - click.echo(json.dumps(value, indent=2)) - - -@cli.command() -@click.option('--path', required=True) -@click.option('--value', required=True) -@click.pass_context -async def set_value(ctx: click.Context, path: str, value: str) -> None: - client = ctx.find_object(SagemcomClient) - if client is None: - return - try: - value = await client.set_value_by_xpath(path, value) - except Exception as e: - click.echo(e, err=True) - raise click.Abort() - - -@cli.command() -@click.pass_context -async def get_wan_mode(ctx: click.Context) -> None: - await ctx.invoke(get_value, path='Device/Services/BellNetworkCfg/WanMode') - - -@cli.command() -@click.pass_context -async def get_dns(ctx: click.Context) -> None: - await ctx.invoke(get_value, path='Device/DNS') - - -@cli.command() -@click.option('-s', '--dns-servers', required=True, nargs=2, type=IPv4Address) -@click.pass_obj -async def set_dns_servers(client: SagemcomClient, dns_servers: tuple[IPv4Address]) -> None: - try: - forwards = await client.get_value_by_xpath('Device/DNS/Relay/Forwardings') - autos = {forward['uid'] for forward in forwards \ - if forward.keys() >= {'uid', 'Alias', 'Interface', 'Enable'} and \ - forward['Alias'].startswith('IPCP') and \ - forward['Interface'].endswith('[IP_DATA]') and \ - forward['Enable']} - statics = {forward['uid'] for forward in forwards \ - if forward.keys() >= {'uid', 'Alias', 'Interface'} and \ - forward['Alias'].startswith('STATIC') and \ - forward['Interface'].endswith(('[IP_DATA]', '[IP_BR_LAN]'))} - for uid in autos: - await client.set_value_by_xpath(f"Device/DNS/Relay/Forwardings/Forwarding[@uid={uid}]/Enable", False) - for uid, dns_server in zip(statics, dns_servers): - await client.set_value_by_xpath(f"Device/DNS/Relay/Forwardings/Forwarding[@uid={uid}]/DNSServer", dns_server) - await client.set_value_by_xpath( - f"Device/DNS/Relay/Forwardings/Forwarding[@uid={uid}]/Interface", - 'Device/IP/Interfaces/Interface[IP_BR_LAN]' if dns_server.is_private else 'Device/IP/Interfaces/Interface[IP_DATA]' - ) - await client.set_value_by_xpath(f"Device/DNS/Relay/Forwardings/Forwarding[@uid={uid}]/Enable", True) - except Exception as e: - click.echo(e, err=True) - raise click.Abort() - - -@cli.command() -@click.option('-r', '--radios', multiple=True) -@click.pass_obj -async def disable_wifi_radios(client: SagemcomClient, radios: tuple[str] | list[str]) -> None: - try: - value = await client.get_value_by_xpath('Device/WiFi/Radios') - active_radios = {radio['Alias'] for radio in value if radio.keys() >= {'Alias', 'Enable'} and \ - radio['Enable']} - if not len(radios): - radios = click.prompt('Choose radio', type=click.Choice(active_radios), show_choices=True), - invalid_radios = set(radios) - active_radios - if len(invalid_radios): - raise click.BadParameter("Invalid radio(s): {0}".format(", ".join(invalid_radios))) - disable_radios = set(radios) & active_radios - for alias in disable_radios: - await client.set_value_by_xpath(f"Device/WiFi/Radios/Radio[Alias='{alias}']/Enable", False) - except Exception as e: - click.echo(e, err=True) - raise click.Abort() - - -def validate_mac_address(ctx: click.Context, param: click.Parameter, value: str) -> str: - result = value.upper() - if not re.match(r"([0-9A-F]{2}:){5}[0-9A-F]{2}$", result): - raise click.BadParameter('Invalid MAC address', ctx, param) - return result - - -@cli.command() -@click.option('-m', '--mac-address', callback=validate_mac_address, prompt='MAC Address') -@click.pass_obj -async def enable_advanced_dmz(client: SagemcomClient, mac_address: str) -> None: - try: - await client.set_value_by_xpath('Device/Services/BellNetworkCfg/AdvancedDMZ/Enable', False) - await client.set_value_by_xpath('Device/Services/BellNetworkCfg/AdvancedDMZ/AdvancedDMZhost', mac_address) - await client.set_value_by_xpath('Device/Services/BellNetworkCfg/AdvancedDMZ/Enable', True) - except Exception as e: - click.echo(e, err=True) - raise click.Abort() - - -if __name__ == '__main__': - config = dict() - if os.path.isfile('config.yaml'): - with open('config.yaml') as f: - config = yaml.safe_load(f) - cli(default_map=config, _anyio_backend='asyncio') - diff --git a/xmo/__init__.py b/xmo/__init__.py new file mode 100644 index 0000000..b8023d8 --- /dev/null +++ b/xmo/__init__.py @@ -0,0 +1 @@ +__version__ = '0.0.1' diff --git a/xmo/__main__.py b/xmo/__main__.py new file mode 100644 index 0000000..5c5863c --- /dev/null +++ b/xmo/__main__.py @@ -0,0 +1,19 @@ +import os +import xmo.dns +import xmo.dmz +import xmo.wifi +import yaml +from . import xmo + + +def main(): + config = dict() + if os.path.isfile('config.yaml'): + with open('config.yaml') as f: + config = yaml.safe_load(f) + xmo.cli(default_map=config, _anyio_backend='asyncio') + + +if __name__ == '__main__': + main() + diff --git a/xmo/dmz.py b/xmo/dmz.py new file mode 100644 index 0000000..2b6a7fa --- /dev/null +++ b/xmo/dmz.py @@ -0,0 +1,28 @@ +import asyncclick as click +import re +from sagemcom_api.client import SagemcomClient +from . import xmo + +def _validate_mac_address(ctx: click.Context, param: click.Parameter, value: str) -> str: + result = value.upper() + if not re.match(r"([0-9A-F]{2}:){5}[0-9A-F]{2}$", result): + raise click.BadParameter('Invalid MAC address', ctx, param) + return result + +@xmo.cli.command() +@click.option('-m', '--mac-address', callback=_validate_mac_address, prompt='MAC Address') +@click.pass_obj +async def enable_advanced_dmz(client: SagemcomClient, mac_address: str) -> None: + try: + await client.set_value_by_xpath('Device/Services/BellNetworkCfg/AdvancedDMZ/Enable', False) + await client.set_value_by_xpath('Device/Services/BellNetworkCfg/AdvancedDMZ/AdvancedDMZhost', mac_address) + await client.set_value_by_xpath('Device/Services/BellNetworkCfg/AdvancedDMZ/Enable', True) + except Exception as e: + click.echo(e, err=True) + raise click.Abort() + +@xmo.cli.command() +@click.pass_context +async def disable_advanced_dmz(ctx: click.Context): + await ctx.invoke(set_value, path='Device/Services/BellNetworkCfg/AdvancedDMZ/Enable', value=False) + diff --git a/xmo/dns.py b/xmo/dns.py new file mode 100644 index 0000000..9434e46 --- /dev/null +++ b/xmo/dns.py @@ -0,0 +1,33 @@ +import asyncclick as click +from ipaddress import IPv4Address +from sagemcom_api.client import SagemcomClient +from . import xmo + +@xmo.cli.command() +@click.option('-s', '--dns-servers', required=True, nargs=2, type=IPv4Address) +@click.pass_obj +async def set_dns_servers(client: SagemcomClient, dns_servers: tuple[IPv4Address]) -> None: + try: + forwards = await client.get_value_by_xpath('Device/DNS/Relay/Forwardings') + autos = {forward['uid'] for forward in forwards \ + if forward.keys() >= {'uid', 'Alias', 'Interface', 'Enable'} and \ + forward['Alias'].startswith('IPCP') and \ + forward['Interface'].endswith('[IP_DATA]') and \ + forward['Enable']} + statics = {forward['uid'] for forward in forwards \ + if forward.keys() >= {'uid', 'Alias', 'Interface'} and \ + forward['Alias'].startswith('STATIC') and \ + forward['Interface'].endswith(('[IP_DATA]', '[IP_BR_LAN]'))} + for uid in autos: + await client.set_value_by_xpath(f"Device/DNS/Relay/Forwardings/Forwarding[@uid={uid}]/Enable", False) + for uid, dns_server in zip(statics, dns_servers): + await client.set_value_by_xpath(f"Device/DNS/Relay/Forwardings/Forwarding[@uid={uid}]/DNSServer", dns_server) + await client.set_value_by_xpath( + f"Device/DNS/Relay/Forwardings/Forwarding[@uid={uid}]/Interface", + 'Device/IP/Interfaces/Interface[IP_BR_LAN]' if dns_server.is_private else 'Device/IP/Interfaces/Interface[IP_DATA]' + ) + await client.set_value_by_xpath(f"Device/DNS/Relay/Forwardings/Forwarding[@uid={uid}]/Enable", True) + except Exception as e: + click.echo(e, err=True) + raise click.Abort() + diff --git a/xmo/wifi.py b/xmo/wifi.py new file mode 100644 index 0000000..ebd60e2 --- /dev/null +++ b/xmo/wifi.py @@ -0,0 +1,24 @@ +import asyncclick as click +from sagemcom_api.client import SagemcomClient +from . import xmo + +@xmo.cli.command() +@click.option('-r', '--radios', multiple=True) +@click.pass_obj +async def disable_wifi_radios(client: SagemcomClient, radios: tuple[str] | list[str]) -> None: + try: + value = await client.get_value_by_xpath('Device/WiFi/Radios') + active_radios = {radio['Alias'] for radio in value if radio.keys() >= {'Alias', 'Enable'} and \ + radio['Enable']} + if not len(radios): + radios = click.prompt('Choose radio', type=click.Choice(active_radios), show_choices=True), + invalid_radios = set(radios) - active_radios + if len(invalid_radios): + raise click.BadParameter("Invalid radio(s): {0}".format(", ".join(invalid_radios))) + disable_radios = set(radios) & active_radios + for alias in disable_radios: + await client.set_value_by_xpath(f"Device/WiFi/Radios/Radio[Alias='{alias}']/Enable", False) + except Exception as e: + click.echo(e, err=True) + raise click.Abort() + diff --git a/xmo/xmo.py b/xmo/xmo.py new file mode 100644 index 0000000..2243949 --- /dev/null +++ b/xmo/xmo.py @@ -0,0 +1,80 @@ +import asyncclick as click +import json +from aiohttp import ClientSession, ClientTimeout +from aiohttp.connector import TCPConnector +from enum import Enum +from ipaddress import IPv4Address +from sagemcom_api.client import SagemcomClient +from sagemcom_api.enums import EncryptionMethod + + +class EnumChoice(click.Choice): + def __init__(self, enum: Enum, case_sensitive=False): + self.__enum = enum + super().__init__(choices=[item.value for item in enum], case_sensitive=case_sensitive) + + def convert(self, value, param, ctx): + if value is None or isinstance(value, Enum): + return value + converted_str = super().convert(value, param, ctx) + return self.__enum(converted_str) + + +@click.group(chain=True) +@click.option('-H', '--host', default='192.168.2.1', help='Hostname or host IP', type=IPv4Address) +@click.option('-u', '--username', default='admin', help='Administrator username') +@click.option('-p', '--password', required=True, help='Administrator password', prompt=True, hide_input=True) +@click.option('-a', '--auth-method', + default=EncryptionMethod.SHA512, type=EnumChoice(EncryptionMethod), + help='Authentication method') +@click.pass_context +async def cli(ctx: click.Context, host: str, username: str, password: str, auth_method: EncryptionMethod) -> None: + ctx.obj = client = await ctx.with_async_resource( + SagemcomClient(host, username, password, auth_method, + ClientSession( + headers={"User-Agent": "XMO_REMOTE_CLIENT/1.0.0"}, + timeout=ClientTimeout(), + connector=TCPConnector(ssl=True), + ), True + ) + ) + try: + await client.login() + except Exception as e: + click.echo(e, err=True) + raise click.Abort() + + +@cli.command() +@click.option('--path', required=True, multiple=True) +@click.pass_context +async def get_value(ctx: click.Context, path: str) -> None: + client = ctx.find_object(SagemcomClient) + if client is None: + return + for _path in path: + try: + value = await client.get_value_by_xpath(_path) + except Exception as e: + click.echo(e, err=True) + continue + #raise click.Abort() + else: + click.echo(json.dumps(value, indent=2)) + + + +@cli.command() +@click.option('--path', required=True) +@click.option('--value', required=True) +@click.pass_context +async def set_value(ctx: click.Context, path: str, value: str) -> None: + client = ctx.find_object(SagemcomClient) + if client is None: + return + try: + value = await client.set_value_by_xpath(path, value) + except Exception as e: + click.echo(e, err=True) + raise click.Abort() +