From e3ee9d8d5a074bd5a667a89c03825adaf657e533 Mon Sep 17 00:00:00 2001 From: "mark.r.godwin@gmail.com" Date: Sun, 5 Mar 2023 20:37:58 +0000 Subject: [PATCH 1/2] Fix issues with host name format. Add AP Lan port settings --- .vscode/settings.json | 7 +- sample_client.py | 37 ++- src/tplink_omada_client/definitions.py | 38 ++- src/tplink_omada_client/devices.py | 237 ++++++++++++------ src/tplink_omada_client/exceptions.py | 35 ++- src/tplink_omada_client/omadaapiconnection.py | 74 ++++-- src/tplink_omada_client/omadaclient.py | 30 ++- src/tplink_omada_client/omadasiteclient.py | 164 +++++++++--- 8 files changed, 438 insertions(+), 184 deletions(-) diff --git a/.vscode/settings.json b/.vscode/settings.json index 457f44d..0bbca61 100644 --- a/.vscode/settings.json +++ b/.vscode/settings.json @@ -1,3 +1,8 @@ { - "python.analysis.typeCheckingMode": "basic" + "python.analysis.typeCheckingMode": "basic", + + "[python]": { + "editor.defaultFormatter": "ms-python.black-formatter", + "editor.formatOnSave": true + } } \ No newline at end of file diff --git a/sample_client.py b/sample_client.py index 4eeb346..0c2736f 100755 --- a/sample_client.py +++ b/sample_client.py @@ -5,11 +5,13 @@ import sys from pprint import pprint from src.tplink_omada_client.omadaclient import OmadaClient +from src.tplink_omada_client.omadasiteclient import AccessPointPortSettings + async def do_the_magic(url: str, username: str, password: str): - """ Not a real test client. """ + """Not a real test client.""" - async with OmadaClient(url, username, password) as client: + async with OmadaClient(url, username, password, verify_ssl=False) as client: print(f"Found Omada Controller: {await client.get_controller_name()}") @@ -27,21 +29,39 @@ async def do_the_magic(url: str, username: str, password: str): print(f" {len([d for d in devices if d.type == 'switch'])} Switches.") print(f" {len([d for d in devices if d.type == 'gateway'])} Routers.") - #pprint(vars(devices[0])) + aps = [ + await site_client.get_access_point(a) for a in devices if a.type == "ap" + ] + for a in aps: + print(f"Access Point: {a.name}") + if a.name == "Office": + port_status = await site_client.update_access_point_port( + a, "ETH3", AccessPointPortSettings(enable_poe=True) + ) + pprint(vars(port_status)) + port_status = await site_client.update_access_point_port( + a, "ETH3", AccessPointPortSettings(enable_poe=False) + ) + pprint(vars(port_status)) + + # pprint(vars(devices[0])) # Get full info of all switches - switches = [await site_client.get_switch(s) for s in devices if s.type == "switch"] + switches = [ + await site_client.get_switch(s) for s in devices if s.type == "switch" + ] - #pprint(vars(switches[0])) + # pprint(vars(switches[0])) - #ports = await client.get_switch_ports(switches[0]) + # ports = await client.get_switch_ports(switches[0]) port = await site_client.get_switch_port(switches[0], switches[0].ports[4]) print(f"Port index 4: {port.name} Profile: {port.profile_name}") pprint(vars(port)) updated_port = await site_client.update_switch_port( - switches[0], port, new_name="Port5") + switches[0], port, new_name="Port5" + ) pprint(vars(updated_port)) profiles = await site_client.get_port_profiles() @@ -49,8 +69,9 @@ async def do_the_magic(url: str, username: str, password: str): print("Done.") + def main(): - """ Basic sample test client. """ + """Basic sample test client.""" if len(sys.argv) < 2 or len(sys.argv) > 4: print("Usage: client [username] [password]") return diff --git a/src/tplink_omada_client/definitions.py b/src/tplink_omada_client/definitions.py index a792f71..6bccb3e 100644 --- a/src/tplink_omada_client/definitions.py +++ b/src/tplink_omada_client/definitions.py @@ -2,11 +2,13 @@ from enum import IntEnum + class DeviceStatus(IntEnum): - """ Known status codes for devices. """ + """Known status codes for devices.""" + DISCONNECTED = 0 DISCONNECTED_MIGRATING = 1 - PROVISIONING = 10 + PROVISIONING = 10 CONFIGURING = 11 UPGRADING = 12 REBOOTING = 13 @@ -29,53 +31,69 @@ class DeviceStatus(IntEnum): ISOLATED = 40 ISOLATED_MIGRATING = 41 + class DeviceStatusCategory(IntEnum): - """ Known status categories for devices """ + """Known status categories for devices""" + DISCONNECTED = 0 CONNECTED = 1 PENDING = 2 HEARTBEAT_MISSED = 3 ISOLATED = 4 + class PortType(IntEnum): - """ Known types of switch port. """ + """Known types of switch port.""" + COPPER = 1 COMBO = 2 SFP = 3 + class LinkStatus(IntEnum): - """ Known link statuses. """ + """Known link statuses.""" + LINK_UP = 0 LINK_DOWN = 2 + class LinkSpeed(IntEnum): - """ Known link speeds. """ + """Known link speeds.""" + SPEED_AUTO = 0 SPEED_10_MBPS = 1 SPEED_100_MBPS = 2 SPEED_1_GBPS = 3 SPEED_10_GBPS = 4 + class LinkDuplex(IntEnum): - """ Known link duplex modes """ + """Known link duplex modes""" + AUTO = 0 HALF = 1 FULL = 2 + class Eth802Dot1X(IntEnum): - """ 802.1x auth modes. """ + """802.1x auth modes.""" + FORCE_UNAUTHORIZED = 0 FORCE_AUTHORIZED = 1 AUTO = 2 + class BandwidthControl(IntEnum): - """ Modes of bandwidth control. """ + """Modes of bandwidth control.""" + OFF = 0 RATE_LIMIT = 1 STORM_CONTROL = 2 + class PoEMode(IntEnum): - """ Settings for PoE policy. """ + """Settings for PoE policy.""" + DISABLED = 0 ENABLED = 1 USE_DEVICE_SETTINGS = 2 diff --git a/src/tplink_omada_client/devices.py b/src/tplink_omada_client/devices.py index 0e8f0f0..4461945 100644 --- a/src/tplink_omada_client/devices.py +++ b/src/tplink_omada_client/devices.py @@ -3,7 +3,7 @@ APs, Switches and Routers """ -from typing import (Any, Dict, List, Optional) +from typing import Any, Dict, List, Optional from .definitions import ( BandwidthControl, DeviceStatus, @@ -12,219 +12,240 @@ LinkSpeed, LinkStatus, PoEMode, - PortType + PortType, ) class OmadaDevice: - """ Details of a device connected to the controller """ + """Details of a device connected to the controller""" def __init__(self, data: Dict[str, Any]): self._data = data @property def type(self) -> str: - """ The type of the device. Its value can be "ap", "gateway", and "switch". """ + """The type of the device. Its value can be "ap", "gateway", and "switch".""" return self._data["type"] @property def mac(self) -> str: - """ The MAC address of the device.""" + """The MAC address of the device.""" return self._data["mac"] @property def name(self) -> str: - """ The device name. """ + """The device name.""" return self._data["name"] @property def model(self) -> str: - """ The device model, such as EAP225. """ + """The device model, such as EAP225.""" return self._data["model"] @property def model_display_name(self) -> str: - """ Model description for front-end display. """ + """Model description for front-end display.""" return self._data["showModel"] @property def status(self) -> DeviceStatus: - """ The status of the device. """ + """The status of the device.""" return self._data["status"] @property def status_category(self) -> DeviceStatusCategory: - """ The high-level status of the device. """ + """The high-level status of the device.""" return self._data["statusCategory"] @property def ip_address(self) -> str: - """ IP address of the device. """ + """IP address of the device.""" return self._data["ip"] @property def display_uptime(self) -> str: - """ Uptime of the device, as a display string """ + """Uptime of the device, as a display string""" return self._data["uptime"] @property def uptime(self) -> int: - """ Uptime of the device, as a display string """ + """Uptime of the device, as a display string""" return self._data["uptimeLong"] + +class OmadaListDevice(OmadaDevice): + """An Omada Device (router, switch, eap) as represented in the device list""" + + @property + def need_upgrade(self) -> bool: + """True, if a firmware upgrade is available for the device.""" + return self._data["needUpgrade"] + + @property + def fw_download(self) -> bool: + """True, if a firmware upgrade is being downloaded.""" + return self._data["fwDownload"] + + class OmadaLink: - """ Up/Downlink connection from a switch/ap device. """ + """Up/Downlink connection from a switch/ap device.""" + def __init__(self, data: Dict[str, Any]): self._data = data @property def mac(self) -> str: - """ The MAC of the linked device. """ + """The MAC of the linked device.""" return self._data["mac"] @property def name(self) -> str: - """ The name of the linked device. """ + """The name of the linked device.""" return self._data["name"] @property def type(self) -> str: - """ The type of device linked to. """ + """The type of device linked to.""" return self._data["type"] @property def port(self) -> int: - """ The port's number. """ + """The port's number.""" return self._data["port"] class OmadaDownlink(OmadaLink): - """ Downlink connection from a switch/ap port. """ + """Downlink connection from a switch/ap port.""" + @property def type(self) -> str: - """ The type of device downlinked to. """ + """The type of device downlinked to.""" return "ap" @property def model(self) -> str: - """ The model name of device linked to. """ + """The model name of device linked to.""" return self._data["model"] class OmadaUplink(OmadaLink): - """ Uplink connection from a switch/ap device. """ + """Uplink connection from a switch/ap device.""" class OmadaPortStatus: - """ Status information for a switch port. """ + """Status information for a switch port.""" + def __init__(self, data: Dict[str, Any]): self._data = data @property def link_status(self) -> LinkStatus: - """ Port's link status. """ + """Port's link status.""" return self._data["linkStatus"] @property def link_speed(self) -> LinkSpeed: - """ Port's link speed. """ + """Port's link speed.""" return self._data["linkSpeed"] @property def poe_active(self) -> bool: - """ Is the port powering a PoE device? """ + """Is the port powering a PoE device?""" return self._data["poe"] @property def poe_power(self) -> float: - """ Power (W) supplied over PoE. """ + """Power (W) supplied over PoE.""" if "poePower" in self._data: return self._data["poePower"] return 0.0 @property def bytes_tx(self) -> int: - """ Number of bytes transmitted by the port. """ + """Number of bytes transmitted by the port.""" return self._data["tx"] @property def bytes_rx(self) -> int: - """ Number of bytes received by the port. """ + """Number of bytes received by the port.""" return self._data["rx"] @property def stp_discarding(self) -> bool: - """ Stp blocking status in spanning tree. """ + """Stp blocking status in spanning tree.""" return self._data["stpDiscarding"] + class OmadaSwitchPort: - """ Port on a switch/gateway device. """ + """Port on a switch/gateway device.""" + def __init__(self, data: Dict[str, Any]): self._data = data @property def port(self) -> int: - """ The port's number. """ + """The port's number.""" return self._data["port"] @property def name(self) -> str: - """ The device name. """ + """The device name.""" return self._data["name"] @property def profile_id(self) -> str: - """ ID of the port's config profile. """ + """ID of the port's config profile.""" return self._data["profileId"] @property def type(self) -> PortType: - """ The type of the port. """ + """The type of the port.""" return self._data["type"] @property def operation(self) -> str: - """ Port config: switching, mirroring or aggregating. """ + """Port config: switching, mirroring or aggregating.""" return self._data["operation"] @property def is_disabled(self) -> bool: - """ Is the port disabled? """ + """Is the port disabled?""" return self._data["disable"] @property def port_status(self) -> OmadaPortStatus: - """ Status of the port. """ + """Status of the port.""" return OmadaPortStatus(self._data["portStatus"]) class OmadaSwitchDeviceCaps: - """ Capabilities of a switch. """ + """Capabilities of a switch.""" + def __init__(self, data: Dict[str, Any]): self._data = data @property def poe_ports(self) -> int: - """ Number of PoE ports supported. """ + """Number of PoE ports supported.""" return self._data["poePortNum"] @property def supports_poe(self) -> bool: - """ Is PoE supported. """ + """Is PoE supported.""" return self._data["poeSupport"] @property def supports_bt(self) -> bool: - """ Is BT supported. """ + """Is BT supported.""" return self._data["supportBt"] class OmadaSwitch(OmadaDevice): - """ Details of a switch connected to the controller. """ + """Details of a switch connected to the controller.""" @property def number_of_ports(self) -> int: - """ The number of ports on the switch. """ + """The number of ports on the switch.""" if "portNum" in self._data: return self._data["portNum"] # So much for the docs @@ -232,12 +253,12 @@ def number_of_ports(self) -> int: @property def ports(self) -> List[OmadaSwitchPort]: - """ List of ports attached to the switch. """ + """List of ports attached to the switch.""" return [OmadaSwitchPort(p) for p in self._data["ports"]] @property def uplink(self) -> Optional[OmadaUplink]: - """ Uplink device for this switch. """ + """Uplink device for this switch.""" uplink = self._data["uplink"] if uplink is None: return None @@ -245,81 +266,130 @@ def uplink(self) -> Optional[OmadaUplink]: @property def downlink(self) -> List[OmadaDownlink]: - """ Downlink devices attached to switch. """ + """Downlink devices attached to switch.""" if "downlinkList" in self._data: return [OmadaDownlink(d) for d in self._data["downlinkList"]] return [] @property def device_capabilities(self) -> OmadaSwitchDeviceCaps: - """ Capabilities of the switch. """ + """Capabilities of the switch.""" return OmadaSwitchDeviceCaps(self._data["devCap"]) + +class OmadaAccesPointLanPortSettings: + """A LAN port on an access point.""" + + def __init__(self, data: Dict[str, Any]): + self._data = data + + @property + def port_name(self) -> str: + """Name of the port - can't be edited""" + return self._data["lanPort"] + + @property + def supports_vlan(self) -> bool: + """True if the port supports VLAN tagging""" + return self._data["supportVlan"] + + @property + def local_vlan_enable(self) -> bool: + """True if VLAN tagging is enabled for the port explicitly""" + return self._data["localVlanEnable"] + + @property + def local_vlan_id(self) -> int: + """VLAN ID for this port""" + return self._data["localVlanId"] + + @property + def supports_poe(self) -> bool: + """True if the port supports PoE output""" + return self._data["supportPoe"] + + @property + def poe_enable(self) -> bool: + """ + True to enable PoE. + + WARNING: Do not enable PoE for EAPs powered from another EAP + """ + return self._data["poeOutEnable"] + + class OmadaAccessPoint(OmadaDevice): - """ Details of an Access Point connected to the controller. """ + """Details of an Access Point connected to the controller.""" @property def wireless_linked(self) -> bool: - """ True, if the AP is connected wirelessley. """ + """True, if the AP is connected wirelessley.""" return self._data["wirelessLinked"] @property def supports_5g(self) -> bool: - """ True if 5G wifi is supported """ + """True if 5G wifi is supported""" return self._data["deviceMisc"]["support5g"] @property def supports_5g2(self) -> bool: - """ True if 5G2 wifi is supported """ + """True if 5G2 wifi is supported""" return self._data["deviceMisc"]["support5g2"] @property def supports_6g(self) -> bool: - """ True if Wifi 6 is supported """ + """True if Wifi 6 is supported""" return self._data["deviceMisc"]["support6g"] @property def supports_11ac(self) -> bool: - """ True if PoE is supported """ + """True if PoE is supported""" return self._data["deviceMisc"]["support11ac"] @property def supports_mesh(self) -> bool: - """ True if mesh networking is supported """ + """True if mesh networking is supported""" return self._data["deviceMisc"]["supportMesh"] + @property + def lan_port_settings(self) -> List[OmadaAccesPointLanPortSettings]: + """Settings for the LAN ports on the access point""" + return [ + OmadaAccesPointLanPortSettings(p) for p in self._data["lanPortSettings"] + ] + class OmadaSwitchPortDetails(OmadaSwitchPort): - """ Full details of a port on a switch. """ + """Full details of a port on a switch.""" @property def port_id(self) -> str: - """ The ID of the port """ + """The ID of the port""" return self._data["id"] @property def max_speed(self) -> LinkSpeed: - """ The max speed of the port. """ + """The max speed of the port.""" return self._data["maxSpeed"] @property def profile_name(self) -> str: - """ Name of the port's config profile """ + """Name of the port's config profile""" return self._data["profileName"] @property def has_profile_override(self) -> bool: - """ True if the port's config profile has been overridden. """ + """True if the port's config profile has been overridden.""" return self._data["profileOverrideEnable"] @property def poe_mode(self) -> PoEMode: - """ PoE config for this port. """ + """PoE config for this port.""" return self._data["poe"] @property def bandwidth_limit_mode(self) -> BandwidthControl: - """ Type of bandwidth control applied. """ + """Type of bandwidth control applied.""" return self._data["bandWidthCtrlType"] # "bandCtrl": { @@ -343,101 +413,104 @@ def bandwidth_limit_mode(self) -> BandwidthControl: @property def eth_802_1x_control(self) -> Eth802Dot1X: - """ 802.1x Auth mode """ + """802.1x Auth mode""" return self._data["dot1x"] @property def lldp_med_enabled(self) -> bool: - """ LLDP Mode """ + """LLDP Mode""" return self._data["lldpMedEnable"] @property def topology_notify_enabled(self) -> bool: - """ Topology notify mode """ + """Topology notify mode""" return self._data["topoNotifyEnable"] @property def spanning_tree_enabled(self) -> bool: - """ Spanning tree loopback control """ + """Spanning tree loopback control""" return self._data["spanningTreeEnable"] @property def loopback_detect_enabled(self) -> bool: - """ Loopback detection """ + """Loopback detection""" return self._data["loopbackDetectEnable"] @property def port_isolation_enabled(self) -> bool: - """ Port isolation (Danger!) """ + """Port isolation (Danger!)""" return self._data["portIsolationEnable"] class OmadaPortProfile: - """ Definition of a switch port configuration profile. """ + """Definition of a switch port configuration profile.""" + def __init__(self, data: Dict[str, Any]): self._data = data @property def profile_id(self) -> str: - """ ID of this profile. """ + """ID of this profile.""" return self._data["id"] @property def site(self) -> str: - """ Site which this profile is valid for. """ + """Site which this profile is valid for.""" return self._data["site"] @property def name(self) -> str: - """ Name of the profile. """ + """Name of the profile.""" return self._data["name"] @property def poe_mode(self) -> PoEMode: - """ PoE mode. """ + """PoE mode.""" return self._data["poe"] @property def bandwidth_limit_mode(self) -> BandwidthControl: - """ Type of bandwidth control applied. """ + """Type of bandwidth control applied.""" return self._data["bandWidthCtrlType"] @property def eth_802_1x_control(self) -> Eth802Dot1X: - """ 802.1x Auth mode """ + """802.1x Auth mode""" return self._data["dot1x"] @property def lldp_med_enabled(self) -> bool: - """ LLDP Mode """ + """LLDP Mode""" return self._data["lldpMedEnable"] @property def topology_notify_enabled(self) -> bool: - """ Topology notify mode """ + """Topology notify mode""" return self._data["topoNotifyEnable"] @property def spanning_tree_enabled(self) -> bool: - """ Spanning tree loopback control """ + """Spanning tree loopback control""" return self._data["spanningTreeEnable"] @property def loopback_detect_enabled(self) -> bool: - """ Loopback detection """ + """Loopback detection""" return self._data["loopbackDetectEnable"] @property def port_isolation_enabled(self) -> bool: - """ Port isolation (Danger!) """ + """Port isolation (Danger!)""" return self._data["portIsolationEnable"] + class OmadaInterfaceDetails: - """ Basic UI Information about controller. """ + """Basic UI Information about controller.""" + def __init__(self, data: Dict[str, Any]): self._data = data @property def controller_name(self) -> str: - """ Display name of the controller. """ + """Display name of the controller.""" return self._data["controllerName"] diff --git a/src/tplink_omada_client/exceptions.py b/src/tplink_omada_client/exceptions.py index daf8638..1cf723d 100644 --- a/src/tplink_omada_client/exceptions.py +++ b/src/tplink_omada_client/exceptions.py @@ -1,17 +1,32 @@ """ Exceptions that the library might throw. """ + class OmadaClientException(Exception): - """ Base for all exceptions raised by the library. """ + """Base for all exceptions raised by the library.""" + class RequestFailed(OmadaClientException): - """ Generic rejection of any command by the controller. """ + """Generic rejection of any command by the controller.""" + def __init__(self, error_code: int, msg: str): self._error_code = error_code self._msg = msg super().__init__(f"Omada controller responded '{msg}' ({error_code})") + class LoginFailed(RequestFailed): - """ Username/Password failure. """ + """Username/Password failure.""" + + +class LoginSessionClosed(OmadaClientException): + """ + The login token isn't valid any more. + + If this happens immediately after logging on, and you are using IP addresses to contact the controller + then make sure you supply a ClientSession that has an unsafe CookieJar, or the login session cookies + won't work. + """ + class UnsupportedControllerVersion(OmadaClientException): """ @@ -19,18 +34,22 @@ class UnsupportedControllerVersion(OmadaClientException): Only controller versions 5.0 and later are supported. """ + def __init__(self, version): super().__init__(f"Unsupported Omada controller version {version} found.") + class SiteNotFound(OmadaClientException): - """ The specified site cannot be found on the Controller. """ + """The specified site cannot be found on the Controller.""" + class ConnectionFailed(OmadaClientException): - """ Connection to Omada controller failed at the network level. """ + """Connection to Omada controller failed at the network level.""" + class BadControllerUrl(OmadaClientException): - """ URL of controller could not be resolved. """ + """URL of controller could not be resolved.""" + class InvalidDevice(OmadaClientException): - """ Device type isn't valid for this operation. """ - \ No newline at end of file + """Device type isn't valid for this operation.""" diff --git a/src/tplink_omada_client/omadaapiconnection.py b/src/tplink_omada_client/omadaapiconnection.py index e061f65..d619a4f 100644 --- a/src/tplink_omada_client/omadaapiconnection.py +++ b/src/tplink_omada_client/omadaapiconnection.py @@ -3,11 +3,19 @@ import time from typing import Any, Optional, Tuple -from aiohttp import client_exceptions +import re +from urllib.parse import urlsplit, urljoin +from aiohttp import client_exceptions, CookieJar from aiohttp.client import ClientSession -from .exceptions import (BadControllerUrl, ConnectionFailed, LoginFailed, - RequestFailed, UnsupportedControllerVersion) +from .exceptions import ( + BadControllerUrl, + ConnectionFailed, + LoginFailed, + LoginSessionClosed, + RequestFailed, + UnsupportedControllerVersion, +) class OmadaApiConnection: @@ -20,14 +28,20 @@ class OmadaApiConnection: _last_logon: float def __init__( - self, - url: str, - username: str, - password: str, - websession: Optional[ClientSession] = None, - verify_ssl=True + self, + url: str, + username: str, + password: str, + websession: Optional[ClientSession] = None, + verify_ssl=True, ): + if not url.lower().startswith(("http://", "https://")): + url = "https://" + url + url_parts = urlsplit(url, "https://") + self._url = url_parts.geturl() + self._host = url_parts.hostname or "" + self._url = url self._username = username self._password = password @@ -38,7 +52,13 @@ def __init__( async def _get_session(self) -> ClientSession: if self._session is None: self._own_session = True - self._session = ClientSession() + jar = ( + None + if re.fullmatch(r"\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}", self._host) + is None + else CookieJar(unsafe=True) + ) + self._session = ClientSession(cookie_jar=jar) return self._session async def __aenter__(self): @@ -75,7 +95,9 @@ async def login(self) -> str: self._controller_version = version auth = {"username": self._username, "password": self._password} - response = await self._do_request("post", self.format_url("login"), payload=auth) + response = await self._do_request( + "post", self.format_url("login"), payload=auth + ) self._csrf_token = response["token"] self._last_logon = time.time() @@ -96,23 +118,23 @@ async def _check_login(self) -> bool: if logged_in: self._last_logon = time.time() return logged_in - except: # pylint: disable=bare-except + except: # pylint: disable=bare-except return False async def _get_controller_info(self) -> Tuple[str, str]: """Get Omada controller version and Id (unauthenticated).""" - response = await self._do_request("get", f"{self._url}/api/info") + response = await self._do_request("get", urljoin(self._url, "/api/info")) - return (response["controllerVer"],response["omadacId"]) + return (response["controllerVer"], response["omadacId"]) - def format_url(self, end_point:str, site:Optional[str]=None) -> str: + def format_url(self, end_point: str, site: Optional[str] = None) -> str: """Get a REST url for the controller action""" if site: end_point = f"sites/{site}/{end_point}" - return f"{self._url}/{self._controller_id}/api/v2/{end_point}" + return urljoin(self._url, f"/{self._controller_id}/api/v2/{end_point}") async def request(self, method: str, url: str, params=None, payload=None) -> Any: """Perform a request specific to the controlller, with authentication""" @@ -122,7 +144,9 @@ async def request(self, method: str, url: str, params=None, payload=None) -> Any return await self._do_request(method, url, params=params, payload=payload) - async def _do_request(self, method: str, url: str, params=None, payload=None) -> Any: + async def _do_request( + self, method: str, url: str, params=None, payload=None + ) -> Any: """Perform a request on the controller, and unpack the response.""" session = await self._get_session() @@ -135,12 +159,12 @@ async def _do_request(self, method: str, url: str, params=None, payload=None) -> try: async with session.request( - method, - url, - params=params, - headers=headers, - json=payload, - ssl=self._verify_ssl, + method, + url, + params=params, + headers=headers, + json=payload, + ssl=self._verify_ssl, ) as response: if response.status != 200: @@ -150,6 +174,10 @@ async def _do_request(self, method: str, url: str, params=None, payload=None) -> raise RequestFailed(response.status, "HTTP Request Error") + # If something goes wrong with the login session, Omada requests return "success", and a login page. :/ + if response.content_type != "application/json": + raise LoginSessionClosed() + content = await response.json() self._check_application_errors(content) diff --git a/src/tplink_omada_client/omadaclient.py b/src/tplink_omada_client/omadaclient.py index 2c15494..094fd4b 100644 --- a/src/tplink_omada_client/omadaclient.py +++ b/src/tplink_omada_client/omadaclient.py @@ -1,5 +1,5 @@ """ Simple Http client for Omada controller REST api. """ -from typing import (NamedTuple, Optional, Union) +from typing import NamedTuple, Optional, Union from aiohttp.client import ClientSession from .omadasiteclient import OmadaSiteClient @@ -13,11 +13,14 @@ OmadaInterfaceDetails, ) + class OmadaSite(NamedTuple): """Identifies a site controlled by the controller.""" + name: str id: str + class OmadaClient: """ Simple client for Omada controller API @@ -25,13 +28,14 @@ class OmadaClient: Provides a very limited subset of the API documented in the 'Omada_SDN_Controller_V5.0.15 API Document' """ + def __init__( - self, - url: str, - username: str, - password: str, - websession: Optional[ClientSession] = None, - verify_ssl=True, + self, + url: str, + username: str, + password: str, + websession: Optional[ClientSession] = None, + verify_ssl=True, ): self._api = OmadaApiConnection(url, username, password, websession, verify_ssl) @@ -55,8 +59,10 @@ async def login(self) -> str: return await self._api.login() async def get_controller_name(self) -> str: - """ Get the display name of the Omada controller. """ - result = await self._api.request("get", self._api.format_url("maintenance/uiInterface")) + """Get the display name of the Omada controller.""" + result = await self._api.request( + "get", self._api.format_url("maintenance/uiInterface") + ) return OmadaInterfaceDetails(result).controller_name @@ -64,14 +70,12 @@ async def get_sites(self) -> list[OmadaSite]: """Get basic list of sites the user can see""" response = await self._api.request("get", self._api.format_url("users/current")) - sites = [ - OmadaSite(s["name"], s["key"]) for s in response["privilege"]["sites"] - ] + sites = [OmadaSite(s["name"], s["key"]) for s in response["privilege"]["sites"]] return sites async def get_site_client(self, site: Union[str, OmadaSite]) -> OmadaSiteClient: """Get a client that can query the specified Omada site.""" - if isinstance(site,OmadaSite): + if isinstance(site, OmadaSite): site_id = site.id else: site_id = await self._get_site_id(site) diff --git a/src/tplink_omada_client/omadasiteclient.py b/src/tplink_omada_client/omadasiteclient.py index 99c8dde..9788ecb 100644 --- a/src/tplink_omada_client/omadasiteclient.py +++ b/src/tplink_omada_client/omadasiteclient.py @@ -4,11 +4,14 @@ from .omadaapiconnection import OmadaApiConnection from .devices import ( + OmadaAccessPoint, OmadaDevice, + OmadaListDevice, OmadaPortProfile, OmadaSwitch, OmadaSwitchPort, - OmadaSwitchPortDetails + OmadaSwitchPortDetails, + OmadaAccesPointLanPortSettings, ) from .exceptions import ( @@ -17,6 +20,7 @@ from .definitions import BandwidthControl, Eth802Dot1X, LinkDuplex, LinkSpeed, PoEMode + class SwitchPortOverrides: """ Overrides that can be applied to a switch port. @@ -26,7 +30,9 @@ class SwitchPortOverrides: we can't just override a single profile setting. Therefore, you may need to initialise all of these parameters to avoid overwriting settings. """ - def __init__(self, + + def __init__( + self, enable_poe: bool = True, dot1x_mode: Eth802Dot1X = Eth802Dot1X.FORCE_AUTHORIZED, duplex: LinkDuplex = LinkDuplex.AUTO, @@ -34,7 +40,7 @@ def __init__(self, lldp_med_enable: bool = True, loopback_detect: bool = True, spanning_tree_enable: bool = False, - port_isolation: bool = False + port_isolation: bool = False, ): self.enable_poe = enable_poe self.dot1x_mode = dot1x_mode @@ -46,36 +52,78 @@ def __init__(self, self.port_isolation = port_isolation +class AccessPointPortSettings: + """ + Settings that can be applied to network ports on access points + + Specify the values you want to modify. The remaining values will be unaffected + """ + + def __init__( + self, + enable_poe: Optional[bool] = None, + vlan_enable: Optional[bool] = None, + vlan_id: Optional[int] = None, + ): + self.enable_poe = enable_poe + self.vlan_enable = vlan_enable + self.vlan_id = vlan_id class OmadaSiteClient: """Client for querying an Omada site's devices.""" - def __init__( - self, - site_id: str, - api: OmadaApiConnection): + def __init__(self, site_id: str, api: OmadaApiConnection): self._api = api self._site_id = site_id - async def get_devices(self) -> List[OmadaDevice]: - """ Get the list of devices on the site. """ + async def get_devices(self) -> List[OmadaListDevice]: + """Get the list of devices on the site.""" result = await self._api.request( - "get", - self._api.format_url("devices", self._site_id) + "get", self._api.format_url("devices", self._site_id) ) - return [OmadaDevice(d) - for d in result] + return [OmadaListDevice(d) for d in result] async def get_switches(self) -> List[OmadaSwitch]: - """ Get the list of switches on the site. """ + """Get the list of switches on the site.""" + + return [ + await self.get_switch(d) + for d in await self.get_devices() + if d.type == "switch" + ] + + async def get_access_points(self) -> List[OmadaAccessPoint]: + """Get the list of access points on the site.""" + + return [ + await self.get_access_point(d) + for d in await self.get_devices() + if d.type == "ap" + ] + + async def get_access_point( + self, mac_or_device: Union[str, OmadaDevice] + ) -> OmadaAccessPoint: + """Get an access point by Mac address or Omada device.""" + + if isinstance(mac_or_device, OmadaDevice): + if mac_or_device.type != "ap": + raise InvalidDevice() + mac = mac_or_device.mac + else: + mac = mac_or_device + + result = await self._api.request( + "get", self._api.format_url(f"eaps/{mac}", self._site_id) + ) - return [await self.get_switch(d) for d in await self.get_devices() if d.type == "switch"] + return OmadaAccessPoint(result) async def get_switch(self, mac_or_device: Union[str, OmadaDevice]) -> OmadaSwitch: - """ Get a switch by Mac address or Omada device. """ + """Get a switch by Mac address or Omada device.""" if isinstance(mac_or_device, OmadaDevice): if mac_or_device.type != "switch": @@ -85,17 +133,15 @@ async def get_switch(self, mac_or_device: Union[str, OmadaDevice]) -> OmadaSwitc mac = mac_or_device result = await self._api.request( - "get", - self._api.format_url(f"switches/{mac}", self._site_id) + "get", self._api.format_url(f"switches/{mac}", self._site_id) ) return OmadaSwitch(result) async def get_switch_ports( - self, - mac_or_device: Union[str, OmadaDevice] - ) -> List[OmadaSwitchPortDetails]: - """ Get a switch by Mac address or Omada device. """ + self, mac_or_device: Union[str, OmadaDevice] + ) -> List[OmadaSwitchPortDetails]: + """Get a switch by Mac address or Omada device.""" if isinstance(mac_or_device, OmadaDevice): if mac_or_device.type != "switch": @@ -105,8 +151,7 @@ async def get_switch_ports( mac = mac_or_device result = await self._api.request( - "get", - self._api.format_url(f"switches/{mac}/ports", self._site_id) + "get", self._api.format_url(f"switches/{mac}/ports", self._site_id) ) return [OmadaSwitchPortDetails(p) for p in result] @@ -114,9 +159,9 @@ async def get_switch_ports( async def get_switch_port( self, mac_or_device: Union[str, OmadaDevice], - index_or_port: Union[int, OmadaSwitchPort] - ) -> OmadaSwitchPortDetails: - """ Get a switch by Mac address or Omada device. """ + index_or_port: Union[int, OmadaSwitchPort], + ) -> OmadaSwitchPortDetails: + """Get a switch by Mac address or Omada device.""" if isinstance(mac_or_device, OmadaDevice): if mac_or_device.type != "switch": @@ -131,21 +176,61 @@ async def get_switch_port( port = index_or_port result = await self._api.request( - "get", - self._api.format_url(f"switches/{mac}/ports/{port}", self._site_id) + "get", self._api.format_url(f"switches/{mac}/ports/{port}", self._site_id) ) return OmadaSwitchPortDetails(result) + async def update_access_point_port( + self, + mac_or_device: Union[str, OmadaDevice], + port_name: str, + setting: AccessPointPortSettings, + ) -> OmadaAccesPointLanPortSettings: + """Update the settings for a lan port on the access point.""" + + # Get the latest representation of the acccess point + ap = await self.get_access_point(mac_or_device) + + port_settings = [ + { + "id": port_name, + "lanPort": port_name, + "localVlanEnable": setting.vlan_enable + if setting.vlan_enable is not None + else ps.local_vlan_enable, + "localVlanId": setting.vlan_id + if setting.vlan_id is not None + else ps.local_vlan_id, + "poeOutEnable": setting.enable_poe + if setting.enable_poe is not None and ps.supports_poe + else ps.poe_enable, + } + for ps in ap.lan_port_settings + if ps.port_name == port_name + ] + + payload = {"lanPortSettings": port_settings} + + result = await self._api.request( + "patch", + self._api.format_url(f"eaps/{ap.mac}", self._site_id), + payload=payload, + ) + + updated_ap = OmadaAccessPoint(result) + # The caller probably only cares about the updated port status + return next(p for p in updated_ap.lan_port_settings if p.port_name == port_name) + async def update_switch_port( self, mac_or_device: Union[str, OmadaDevice], index_or_port: Union[int, OmadaSwitchPort], new_name: Optional[str] = None, profile_id: Optional[str] = None, - overrides: Optional[SwitchPortOverrides] = None - ) -> OmadaSwitchPortDetails: - """ Applies an existing profile to a switch on the port """ + overrides: Optional[SwitchPortOverrides] = None, + ) -> OmadaSwitchPortDetails: + """Applies an existing profile to a switch on the port""" if isinstance(mac_or_device, OmadaDevice): if mac_or_device.type != "switch": @@ -162,12 +247,14 @@ async def update_switch_port( payload = { "name": new_name or port.name, "profileId": profile_id or port.profile_id, - "profileOverrideEnable": not overrides is None - } + "profileOverrideEnable": not overrides is None, + } if overrides: payload["operation"] = "switching" payload["bandWidthCtrlType"] = BandwidthControl.OFF - payload["poe"] = PoEMode.ENABLED if overrides.enable_poe else PoEMode.DISABLED + payload["poe"] = ( + PoEMode.ENABLED if overrides.enable_poe else PoEMode.DISABLED + ) payload["dot1x"] = overrides.dot1x_mode payload["duplex"] = overrides.duplex payload["linkSpeed"] = overrides.link_speed @@ -180,18 +267,17 @@ async def update_switch_port( await self._api.request( "patch", self._api.format_url(f"switches/{mac}/ports/{port.port}", self._site_id), - payload = payload + payload=payload, ) # Read back the new port settings return await self.get_switch_port(mac, port) async def get_port_profiles(self) -> List[OmadaPortProfile]: - """ Lists the available switch port profiles that can be applied. """ + """Lists the available switch port profiles that can be applied.""" result = await self._api.request( - "get", - self._api.format_url("setting/lan/profileSummary", self._site_id) + "get", self._api.format_url("setting/lan/profileSummary", self._site_id) ) return [OmadaPortProfile(p) for p in result["data"]] From cd073dd35372cb25470a702e1292b50a6bbdfd63 Mon Sep 17 00:00:00 2001 From: "mark.r.godwin@gmail.com" Date: Sun, 5 Mar 2023 20:44:50 +0000 Subject: [PATCH 2/2] Bump version --- README.md | 3 ++- pyproject.toml | 2 +- 2 files changed, 3 insertions(+), 2 deletions(-) diff --git a/README.md b/README.md index f85f4fa..5ffb7c7 100644 --- a/README.md +++ b/README.md @@ -17,7 +17,8 @@ Only a subset of the controller's features are supported: * Within site: * List Devices (APs, Gateways and Switches) * Basic device information - * Port status and PoE configuraton for Switches (<-- What I actually needed) + * Port status and PoE configuraton for Switches + * Lan port configuration for Access Points Tested with OC200 on Omada Controller Version 5.5.7 - 5.7.6. Other versions may not be fully compatible. diff --git a/pyproject.toml b/pyproject.toml index cd8ca95..2b96fff 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "hatchling.build" [project] name = "tplink_omada_client" -version = "1.1.0" +version = "1.1.1" authors = [ { name="Mark Godwin", email="author@example.com" }, ]