diff --git a/pyess/aio_ess.py b/pyess/aio_ess.py index f84098d..5d85a56 100755 --- a/pyess/aio_ess.py +++ b/pyess/aio_ess.py @@ -11,8 +11,19 @@ import aiohttp -from pyess.constants import LOGIN_URL, TIMESYNC_URL, GRAPH_TIMESPANS, GRAPH_DEVICES, GRAPH_PARAMS, \ - GRAPH_TFORMATS, SWITCH_URL, STATE_URLS, BATT_URL +from pyess.constants import ( + LOGIN_URL, + TIMESYNC_URL, + LOGIN_URLS, + TIMESYNC_URLS, + GRAPH_TIMESPANS, + GRAPH_DEVICES, + GRAPH_PARAMS, + GRAPH_TFORMATS, + SWITCH_URL, + STATE_URLS, + BATT_URL, +) class ESSException(Exception): @@ -28,48 +39,159 @@ class ESSAuthException(ESSException): class ESS: @classmethod - async def create(cls, name=None, password=None, ip=None): - ess = cls(name, password, ip) + async def create(cls, name=None, password=None, ip=None, session=None): + """ + Create and log in a new ESS client. + + :param session: Optional pre-built aiohttp ``ClientSession``. + When supplied (typically by tests using ``aiohttp.test_utils``), + the client uses it instead of creating its own. The caller + is then responsible for closing the session. + """ + ess = cls(name, password, ip, session=session) await ess._login() return ess - def __init__(self, name, pw, ip=None): + def __init__(self, name, pw, ip=None, session=None): self.name = name self.pw = pw self.ip = ip self.logged_in = False self.auth_key = None - self.session = aiohttp.ClientSession(connector=aiohttp.TCPConnector(ssl=False), - timeout=aiohttp.ClientTimeout(connect=60,sock_read=60, sock_connect=60, total=180)) + # The endpoint actually answered by the device. Set during _login + # and reused on subsequent attempts so we don't re-probe every + # time. Stays None until the first successful login. + self._login_url = None + self._timesync_url = None + if session is not None: + self.session = session + else: + self.session = aiohttp.ClientSession(connector=aiohttp.TCPConnector(ssl=False), + timeout=aiohttp.ClientTimeout(connect=60,sock_read=60, sock_connect=60, total=180)) async def _login(self, retry=1): """ - Login to the ESS device. Called by __init__ - :return: + Login to the ESS device. Called by __init__. + + Tries each candidate login URL in ``pyess.constants.LOGIN_URLS`` + in order, sticking with the first one that does not return 404. + This keeps the library working on both the old PMS firmware + (``/v1/user/setting/login``) and the new PMS firmware introduced + somewhere between 10.05.6511 and 10.05.7438 + (``/v1/login``). + + :raises ESSAuthException: if every candidate URL returns 200 but + the device still reports ``status: password mismatched``. """ - url = LOGIN_URL.format(self.ip) - logger.info("fetching auth key") - async with self.session.put(url, json={"password": self.pw}) as r: - response_json = await r.json() - if "status" in response_json and response_json["status"] == "password mismatched": - raise ESSAuthException("wrong password") - # r = requests.put(url, json={"password": self.pw}, verify=False, headers={"Content-Type": "application/json"}) - auth_key = response_json["auth_key"] + # If a previous successful login already pinned an endpoint, use + # it directly. Otherwise walk the candidate list. We look the + # list up via ``pyess.constants`` on every call so monkey-patches + # in tests (and any future re-ordering at runtime) take effect. + import pyess.constants as _const + if self._login_url is not None: + login_candidates = [self._login_url] + else: + login_candidates = list(_const.LOGIN_URLS) + + auth_key = None + for url in login_candidates: + logger.info("fetching auth key via %s", url) + async with self.session.put(url, json={"password": self.pw}) as r: + if r.status == 404: + # Endpoint not implemented on this firmware — try + # the next candidate. + logger.info("login endpoint %s returned 404, trying fallback", url) + continue + try: + response_json = await r.json() + except (JSONDecodeError, ContentTypeError) as e: + raise ESSException( + f"unexpected non-JSON response from {url}: {e!r}" + ) from e + + # We got a JSON body. If the device says the password is + # wrong, that is a real auth failure (not a URL mismatch) — + # surface it immediately, do not try other URLs. + if ( + "status" in response_json + and response_json["status"] == "password mismatched" + ): + raise ESSAuthException("wrong password") + + # The current LG firmware returns either ``auth_key`` (old + # build) or ``auth`` (new build). Accept both. + auth_key = response_json.get("auth_key") or response_json.get("auth") + if auth_key is None: + raise ESSException( + f"login response from {url} missing auth_key: {response_json!r}" + ) + + # Pin the working URL for the lifetime of this client. + self._login_url = url + break + else: + # Every candidate returned 404 — the device speaks neither + # old nor new login URL. Surface a clear error instead of + # the historical misleading "wrong password". + raise ESSException( + f"no working login endpoint found; tried: {login_candidates}" + ) + + # TimeSync is best-effort. Both old and new firmware builds + # answer differently: legacy returns 200, the new PMS 10.05.x + # build simply does not implement the endpoint and answers 404 + # to both candidate paths. A 404 here is harmless — the device + # uses its own clock — so we log and continue instead of + # retrying the whole login. Any other failure (5xx, malformed + # body, "status != success") still falls through to the + # historical exponential-backoff retry. timesync_info = { "auth_key": auth_key, "by": "phone", - "date_time": time.strftime("%Y-%m-%d %H:%M:%S", time.gmtime()) + "date_time": time.strftime("%Y-%m-%d %H:%M:%S", time.gmtime()), } - async with self.session.put(TIMESYNC_URL.format(self.ip), json=timesync_info) as r: - try: - response_json = await r.json() - # rt = requests.put(TIMESYNC_URL.format(self.ip), json=timesync_info, verify=False, - # headers={"Content-Type": "application/json"}) - - assert response_json['status'] == 'success', response_json - except (JSONDecodeError, ContentTypeError): + if self._timesync_url is not None: + timesync_candidates = [self._timesync_url] + else: + timesync_candidates = list(_const.TIMESYNC_URLS) + + for ts_url in timesync_candidates: + logger.info("syncing device time via %s", ts_url) + async with self.session.put(ts_url, json=timesync_info) as r: + if r.status == 404: + # Endpoint not implemented on this firmware build — + # harmless. Try the other candidate, and if that + # also 404s, give up silently. + logger.info( + "timesync endpoint %s returned 404 (not implemented on this firmware)", + ts_url, + ) + self._timesync_url = ts_url + continue + try: + response_json = await r.json() + except (JSONDecodeError, ContentTypeError): + time.sleep(retry) + return await self._login(retry=retry * 2) + + if response_json.get("status") == "success": + self._timesync_url = ts_url + break + # Unknown / unexpected body shape — keep the legacy + # behaviour and retry the whole login with backoff. time.sleep(retry) return await self._login(retry=retry * 2) + else: + # All candidates returned 404. Time sync is unavailable on + # this firmware, but we did get an auth_key, so do not loop: + # just log once and accept the device's own clock. + logger.info( + "no working timesync endpoint found on this firmware " + "(tried: %s); continuing without device time sync", + timesync_candidates, + ) + self._timesync_url = None + self.auth_key = auth_key self.logged_in = True return auth_key @@ -77,7 +199,7 @@ async def _login(self, retry=1): async def get_graph(self, device: str, timespan: str, date: datetime.datetime): """ Get the time series data about a device - :param device: the device in question ``["batt", "load", "pv"]`` + :param device: the device in the question ``["batt", "load", "pv"]`` :param timespan: the timespan in question ``["day", "week", "month", "year"]`` :param date: the date specifying the time span. (for week, month and year I guess any date within the timespan of interest will suffice) diff --git a/pyess/constants.py b/pyess/constants.py index 87f138f..dbe11cb 100755 --- a/pyess/constants.py +++ b/pyess/constants.py @@ -1,14 +1,61 @@ #!/usr/bin/python # -*- coding: utf-8 -*- PREFIX = "https://{}/" -LOGIN_URL = f"{PREFIX}v1/user/setting/login" -TIMESYNC_URL = f"{PREFIX}v1/user/setting/timesync" + +# --- Login / TimeSync endpoint handling ------------------------------------- +# +# Older LG ESS Home firmware (PMS <= 10.05.6511, April 2020) exposes: +# PUT /v1/user/setting/login +# PUT /v1/user/setting/timesync +# +# Newer LG ESS Home firmware (PMS >= 10.05.7438, ~mid 2024 onwards) has +# shortened these to: +# PUT /v1/login +# PUT /v1/timesync +# +# `gluap/pyess` historically hard-coded the old paths. On a current device +# this surfaces as "password mismatched" (the device returns 404 for the +# old URL, which the client mis-interprets as a wrong password). +# +# To stay compatible with both firmware generations, ``aio_ess.ESS`` now +# tries the new URL first and falls back to the old URL on 404. The +# constants below expose the full set of candidate URLs in priority +# order; ``aio_ess`` uses them in that order and sticks with whichever +# one the device actually answers. +# +# See https://github.com/gluap/pyess/issues/37 for the original bug +# report and the device-side verification of the new endpoints. + +LOGIN_URL = f"{PREFIX}v1/user/setting/login" # legacy default (kept for back-compat with direct imports) +TIMESYNC_URL = f"{PREFIX}v1/user/setting/timesync" # legacy default + +# Order matters: ESS._login walks this list from index 0 onwards, stopping +# at the first URL that does not return HTTP 404. The first element is +# intentionally the *new* URL so that devices on current firmware are +# served without an extra round trip. +LOGIN_URLS = [ + f"{PREFIX}v1/login", # new firmware (PMS 10.05.7438+) + f"{PREFIX}v1/user/setting/login", # legacy firmware (PMS <= 10.05.6511) +] + +# For TimeSync, the situation is asymmetric: on the new firmware the old +# path 404s *and* the new path 404s as well (the endpoint is simply not +# implemented on the current build). ``aio_ess`` therefore treats a 404 +# here as "device has no time sync — skipping is safe" rather than as a +# hard failure, which prevents the tight login-retry loop that used to +# fill Home Assistant logs. +TIMESYNC_URLS = [ + f"{PREFIX}v1/timesync", # new firmware (currently 404 — harmless) + f"{PREFIX}v1/user/setting/timesync", # legacy firmware (200) +] +# --- /Login / TimeSync endpoint handling ------------------------------------ + STATE_URLS = { "network": f"{PREFIX}v1/user/setting/network", "systeminfo": f"{PREFIX}v1/user/setting/systeminfo", "batt": f"{PREFIX}v1/user/setting/batt", - "home": f"{PREFIX}v1/user/essinfo/home", - "common": f"{PREFIX}v1/user/essinfo/common", + "home": f"{PREFIX}v1/essinfo/home", + "common": f"{PREFIX}v1/essinfo/common", } GRAPH_TIMESPANS = {"day", "week", "month", "year"} GRAPH_DEVICES = {"batt", "load", "pv"} diff --git a/tests/aio_ess_endpoint_fallback_test.py b/tests/aio_ess_endpoint_fallback_test.py new file mode 100644 index 0000000..1ac3cab --- /dev/null +++ b/tests/aio_ess_endpoint_fallback_test.py @@ -0,0 +1,333 @@ +#!/usr/bin/python +# -*- coding: utf-8 -*- +""" +Regression tests for the new-firmware login URL fallback. + +These tests cover the two scenarios reported in +https://github.com/gluap/pyess/issues/37: + + 1. Device runs the new PMS firmware (>= 10.05.7438) which only + exposes the shortened auth endpoints. + 2. Device runs the legacy PMS firmware (<= 10.05.6511) which only + exposes the original long endpoints. + 3. TimeSync returns 404 on the new firmware and must not trigger the + historical retry-loop. + +Test strategy: build a real aiohttp server in-process, monkey-patch +``pyess.constants.LOGIN_URLS`` / ``TIMESYNC_URLS`` to point at the +local server, run the real ``ESS`` client, and assert behaviour. This +exercises the actual fallback logic and HTTP error handling without +needing a cassette or a real LG device on the network. +""" +import pytest +import pytest_asyncio +from aiohttp import web +from aiohttp.test_utils import TestServer, TestClient + +import pyess.constants as pyess_constants +from pyess.aio_ess import ESS, ESSAuthException, ESSException +from pyess.constants import ( + LOGIN_URL, + TIMESYNC_URL, + LOGIN_URLS as ORIGINAL_LOGIN_URLS, + TIMESYNC_URLS as ORIGINAL_TIMESYNC_URLS, +) +from pyess.constants import LOGIN_URLS, TIMESYNC_URLS # re-import for direct use in test_url_lists_priority_order + + +# ---- minimal in-memory "ESS device" apps ---------------------------------- + +def make_new_firmware_app(): + """ + Mimic LG ESS Home PMS >= 10.05.7438: + * /v1/login -> 200 + {auth_key: ...} + * /v1/user/setting/login -> 404 + * /v1/timesync -> 404 (not implemented on this build) + * /v1/user/setting/timesync-> 404 + """ + async def login_new(request): + body = await request.json() + if body.get("password") != "right": + return web.json_response({"status": "password mismatched"}, status=200) + return web.json_response({"auth_key": "auth-new-fw"}) + + async def legacy_login(request): + return web.Response(status=404, text="not here") + + async def any_timesync(request): + return web.Response(status=404, text="timesync not implemented on this firmware") + + async def get_state(request): + return web.json_response({"ok": True}) + + app = web.Application() + app.router.add_put("/v1/login", login_new) + app.router.add_put("/v1/user/setting/login", legacy_login) + app.router.add_put("/v1/timesync", any_timesync) + app.router.add_put("/v1/user/setting/timesync", any_timesync) + app.router.add_post("/v1/user/essinfo/common", get_state) + app.router.add_post("/v1/essinfo/common", get_state) + return app + + +def make_legacy_firmware_app(): + """ + Mimic LG ESS Home PMS <= 10.05.6511: + * /v1/login -> 404 + * /v1/user/setting/login -> 200 + {auth_key: ...} + * /v1/user/setting/timesync-> 200 + {status: success} + * /v1/timesync -> 404 + """ + async def login_new(request): + return web.Response(status=404, text="not implemented on legacy firmware") + + async def legacy_login(request): + body = await request.json() + if body.get("password") != "right": + return web.json_response({"status": "password mismatched"}, status=200) + return web.json_response({"auth_key": "auth-legacy-fw"}) + + async def timesync_new(request): + return web.Response(status=404, text="no") + + async def timesync_legacy(request): + return web.json_response({"status": "success"}) + + async def get_state(request): + return web.json_response({"ok": True}) + + app = web.Application() + app.router.add_put("/v1/login", login_new) + app.router.add_put("/v1/user/setting/login", legacy_login) + app.router.add_put("/v1/timesync", timesync_new) + app.router.add_put("/v1/user/setting/timesync", timesync_legacy) + app.router.add_post("/v1/user/essinfo/common", get_state) + app.router.add_post("/v1/essinfo/common", get_state) + return app + + +def make_wrong_password_app(): + """Both URLs return 200 + password_mismatched; the client must raise + ESSAuthException without trying forever.""" + async def login_any(request): + return web.json_response({"status": "password mismatched"}, status=200) + async def timesync_any(request): + return web.json_response({"status": "success"}) + + app = web.Application() + app.router.add_put("/v1/login", login_any) + app.router.add_put("/v1/user/setting/login", login_any) + app.router.add_put("/v1/timesync", timesync_any) + app.router.add_put("/v1/user/setting/timesync", timesync_any) + return app + + +def make_unreachable_app(): + """Both candidate URLs return 404 — the device speaks neither. The + client must raise a clear ESSException, *not* ESSAuthException.""" + async def none(request): + return web.Response(status=404, text="no") + app = web.Application() + app.router.add_put("/v1/login", none) + app.router.add_put("/v1/user/setting/login", none) + app.router.add_put("/v1/timesync", none) + app.router.add_put("/v1/user/setting/timesync", none) + return app + + +# ---- fixtures ------------------------------------------------------------- + +@pytest_asyncio.fixture +async def redirect_to_http_server(unused_tcp_port): + """ + Start a TestServer on a free TCP port, then monkey-patch + ``pyess.constants.LOGIN_URLS`` and ``TIMESYNC_URLS`` so the client + talks ``http://127.0.0.1:/...`` instead of the production + ``https:///...`` URLs. The caller decides which app the + server runs. + + Yields a callable ``(app) -> (client, port)``; the caller closes the + client when done. + """ + servers = [] # keep references for cleanup + yield unused_tcp_port + # Cleanup: restore URLs and close any open servers + pyess_constants.LOGIN_URLS = list(ORIGINAL_LOGIN_URLS) + pyess_constants.TIMESYNC_URLS = list(ORIGINAL_TIMESYNC_URLS) + for c in servers: + await c.close() + pyess_constants.LOGIN_URLS = list(ORIGINAL_LOGIN_URLS) + pyess_constants.TIMESYNC_URLS = list(ORIGINAL_TIMESYNC_URLS) + + +async def run_with_app(app_factory, unused_tcp_port): + """Spin up a test server, redirect the ESS client at it, and return + the test client + the redirect helper.""" + server = TestServer(app_factory()) + client = TestClient(server) + await client.start_server() + port = server._server.sockets[0].getsockname()[1] + pyess_constants.LOGIN_URLS = [ + f"http://127.0.0.1:{port}/v1/login", + f"http://127.0.0.1:{port}/v1/user/setting/login", + ] + pyess_constants.TIMESYNC_URLS = [ + f"http://127.0.0.1:{port}/v1/timesync", + f"http://127.0.0.1:{port}/v1/user/setting/timesync", + ] + return client + + +# ---- the actual tests ------------------------------------------------------ + +@pytest.mark.asyncio +async def test_new_firmware_login_succeeds(unused_tcp_port_factory): + """PMS 10.05.7438+: /v1/login answers, /v1/user/setting/login 404s.""" + # Bind the test server to the port the factory just reserved so the + # client URL monkey-patch and the server socket agree. + port = unused_tcp_port_factory() + server = TestServer(make_new_firmware_app(), port=port) + client = TestClient(server) + await client.start_server() + pyess_constants.LOGIN_URLS = [ + f"http://127.0.0.1:{port}/v1/login", + f"http://127.0.0.1:{port}/v1/user/setting/login", + ] + pyess_constants.TIMESYNC_URLS = [ + f"http://127.0.0.1:{port}/v1/timesync", + f"http://127.0.0.1:{port}/v1/user/setting/timesync", + ] + try: + ess = await ESS.create("user", "right", "127.0.0.1", session=client.session) + assert ess.logged_in is True + assert ess.auth_key == "auth-new-fw" + assert ess._login_url is not None + assert ess._login_url.endswith("/v1/login") + finally: + pyess_constants.LOGIN_URLS = list(ORIGINAL_LOGIN_URLS) + pyess_constants.TIMESYNC_URLS = list(ORIGINAL_TIMESYNC_URLS) + await client.close() + + +@pytest.mark.asyncio +async def test_legacy_firmware_login_succeeds(unused_tcp_port_factory): + """PMS <= 10.05.6511: /v1/login 404s, /v1/user/setting/login answers.""" + port = unused_tcp_port_factory() + server = TestServer(make_legacy_firmware_app(), port=port) + client = TestClient(server) + await client.start_server() + pyess_constants.LOGIN_URLS = [ + f"http://127.0.0.1:{port}/v1/login", + f"http://127.0.0.1:{port}/v1/user/setting/login", + ] + pyess_constants.TIMESYNC_URLS = [ + f"http://127.0.0.1:{port}/v1/timesync", + f"http://127.0.0.1:{port}/v1/user/setting/timesync", + ] + try: + ess = await ESS.create("user", "right", "127.0.0.1", session=client.session) + assert ess.logged_in is True + assert ess.auth_key == "auth-legacy-fw" + assert ess._login_url.endswith("/v1/user/setting/login") + finally: + pyess_constants.LOGIN_URLS = list(ORIGINAL_LOGIN_URLS) + pyess_constants.TIMESYNC_URLS = list(ORIGINAL_TIMESYNC_URLS) + await client.close() + + +@pytest.mark.asyncio +async def test_wrong_password_raises_auth_exception(unused_tcp_port_factory): + """Both endpoints return 200 + password_mismatched. The library must + raise ESSAuthException, not retry forever, not raise ESSException.""" + port = unused_tcp_port_factory() + server = TestServer(make_wrong_password_app(), port=port) + client = TestClient(server) + await client.start_server() + pyess_constants.LOGIN_URLS = [ + f"http://127.0.0.1:{port}/v1/login", + f"http://127.0.0.1:{port}/v1/user/setting/login", + ] + pyess_constants.TIMESYNC_URLS = [ + f"http://127.0.0.1:{port}/v1/timesync", + f"http://127.0.0.1:{port}/v1/user/setting/timesync", + ] + try: + with pytest.raises(ESSAuthException): + await ESS.create("user", "wrong", "127.0.0.1", session=client.session) + finally: + pyess_constants.LOGIN_URLS = list(ORIGINAL_LOGIN_URLS) + pyess_constants.TIMESYNC_URLS = list(ORIGINAL_TIMESYNC_URLS) + await client.close() + + +@pytest.mark.asyncio +async def test_no_endpoint_found_raises_clear_exception(unused_tcp_port_factory): + """If *every* candidate returns 404, we must get ESSException + explaining which URLs we tried, not the misleading 'wrong password'.""" + port = unused_tcp_port_factory() + server = TestServer(make_unreachable_app(), port=port) + client = TestClient(server) + await client.start_server() + pyess_constants.LOGIN_URLS = [ + f"http://127.0.0.1:{port}/v1/login", + f"http://127.0.0.1:{port}/v1/user/setting/login", + ] + pyess_constants.TIMESYNC_URLS = [ + f"http://127.0.0.1:{port}/v1/timesync", + f"http://127.0.0.1:{port}/v1/user/setting/timesync", + ] + try: + with pytest.raises(ESSException) as exc: + await ESS.create("user", "right", "127.0.0.1", session=client.session) + # The error message must mention the URLs we tried so users can + # debug their network/firmware combination. + assert "/v1/login" in str(exc.value) + assert "/v1/user/setting/login" in str(exc.value) + # And it must *not* claim "wrong password". + assert "wrong password" not in str(exc.value).lower() + finally: + pyess_constants.LOGIN_URLS = list(ORIGINAL_LOGIN_URLS) + pyess_constants.TIMESYNC_URLS = list(ORIGINAL_TIMESYNC_URLS) + await client.close() + + +@pytest.mark.asyncio +async def test_new_firmware_timesync_404_does_not_loop(unused_tcp_port_factory): + """Regression: when TimeSync 404s on new firmware, login must NOT + trigger the historical tight retry loop.""" + port = unused_tcp_port_factory() + server = TestServer(make_new_firmware_app(), port=port) + client = TestClient(server) + await client.start_server() + pyess_constants.LOGIN_URLS = [ + f"http://127.0.0.1:{port}/v1/login", + f"http://127.0.0.1:{port}/v1/user/setting/login", + ] + pyess_constants.TIMESYNC_URLS = [ + f"http://127.0.0.1:{port}/v1/timesync", + f"http://127.0.0.1:{port}/v1/user/setting/timesync", + ] + try: + ess = await ESS.create("user", "right", "127.0.0.1", session=client.session) + # If the bug were still there this would either hang or take + # orders of magnitude longer. + assert ess.auth_key == "auth-new-fw" + # And we should have walked both timesync URLs, found both + # 404, and recorded that no timesync is available — without + # retrying the login. + assert ess._timesync_url is None + finally: + pyess_constants.LOGIN_URLS = list(ORIGINAL_LOGIN_URLS) + pyess_constants.TIMESYNC_URLS = list(ORIGINAL_TIMESYNC_URLS) + await client.close() + + +def test_url_lists_priority_order(): + """The new URL must come first in LOGIN_URLS so current devices are + served without an extra round-trip; the legacy URL is the fallback. + Backward-compat: the module-level LOGIN_URL/TIMESYNC_URL constants + must still exist and point at the legacy URLs.""" + assert LOGIN_URLS[0].endswith("/v1/login") + assert LOGIN_URLS[1].endswith("/v1/user/setting/login") + assert LOGIN_URL.endswith("/v1/user/setting/login") + assert TIMESYNC_URL.endswith("/v1/user/setting/timesync")