Swap to private solarweb access

tests/test_solarweb.py

@@ -0,0 +1,203 @@
+"""Exercise browser authentication with synthetic credentials and responses only."""
+
+from datetime import date
+from io import BytesIO
+import json
+import unittest
+from urllib.error import URLError
+from urllib.parse import parse_qs, urlsplit
+from unittest.mock import Mock
+
+from solar_relay.http import TransportError
+from solar_relay.solarweb import API_SERVER, LOGIN_SERVER, Response, SolarWebClient, decode_chart, validate_url
+
+SYSTEM = "73733052-6f10-47a5-9746-7f7e76ebcb8c"
+DAY = date(2026, 10, 30)
+STAMP = 1793354400000
+PASSWORD = "synthetic password & Unicode é"
+LOGIN_FORM = b'''<form action="/commonauth" method="post">
+<input type="hidden" name="sessionDataKey" value="synthetic-session">
+<input name="usernameUserInput"><input type="password" name="password">
+<input type="hidden" name="disabled" value="unused" disabled></form>'''
+CALLBACK_FORM = b'''<form action="https://www.solarweb.com/Account/ExternalLoginCallback" method="post">
+<input type="hidden" name="state" value="synthetic-state">
+<input type="hidden" name="id_token" value="synthetic-token"></form>'''
+
+
+def chart(series_id="FromGen", points=None, suffix=" W"):
+    return json.dumps({"settings": {"series": [{"id": series_id, "name": "Localized display name",
+                                                "tooltip": {"valueSuffix": suffix},
+                                                "data": points if points is not None else [[STAMP, 1200], [STAMP + 300_000, None]]}]}}).encode()
+
+
+def login_steps():
+    return [(302, b"", LOGIN_SERVER + "/authenticationendpoint/login.do?sessionDataKey=synthetic-session"),
+            (200, LOGIN_FORM, ""), (200, CALLBACK_FORM, ""),
+            (302, b"", API_SERVER + "/"), (200, b"<html>Account</html>", "")]
+
+
+class ScriptedClient(SolarWebClient):
+    def __init__(self, steps):
+        super().__init__("account@example.test", PASSWORD)
+        self.steps, self.requests = list(steps), []
+
+    def _send(self, url, form, referer, as_json):
+        self.requests.append((url, form, referer, as_json))
+        if not self.steps:
+            raise AssertionError("Unexpected request")
+        status, body, location = self.steps.pop(0)
+        return Response(status, url, body, location)
+
+
+class LoginTests(unittest.TestCase):
+    def test_real_account_flow_and_authenticated_charts_reuse_session(self):
+        client = ScriptedClient(login_steps() + [(200, b"Chart", ""), (200, chart(), ""),
+                                                 (200, chart("ToConsumer"), "")])
+        self.assertEqual(client.get_chart(SYSTEM, "production", DAY), {STAMP: 1200})
+        self.assertEqual(client.get_chart(SYSTEM, "consumption", DAY), {STAMP: 1200})
+        self.assertTrue(client.authenticated)
+        self.assertFalse(client.steps)
+        posts = [(url, form) for url, form, _, _ in client.requests if form is not None]
+        self.assertEqual(posts, [(LOGIN_SERVER + "/commonauth", {"sessionDataKey": "synthetic-session",
+                                 "username": "account@example.test", "usernameUserInput": "account@example.test",
+                                 "password": PASSWORD}),
+                                (API_SERVER + "/Account/ExternalLoginCallback",
+                                 {"state": "synthetic-state", "id_token": "synthetic-token"})])
+        for request, view in zip(client.requests[-2:], ["consumption", "production"]):
+            url, form, referer, as_json = request
+            query = parse_qs(urlsplit(url).query)
+            self.assertEqual(query, {"pvSystemId": [SYSTEM], "year": ["2026"], "month": ["10"],
+                                     "day": ["30"], "interval": ["day"], "view": [view]})
+            self.assertIsNone(form)
+            self.assertTrue(as_json)
+            self.assertEqual(urlsplit(referer).path, "/Chart/Chart")
+        for url, _, _, _ in client.requests:
+            self.assertNotIn(PASSWORD, url)
+            self.assertNotIn("GuestLogOn", url)
+            self.assertNotIn("swqapi", url)
+
+    def test_expired_chart_session_is_renewed_once(self):
+        client = ScriptedClient(login_steps() + [(200, b"Chart", ""), (401, b"", "")]
+                                + login_steps() + [(200, b"Chart", ""), (200, chart(), "")])
+        self.assertEqual(client.get_chart(SYSTEM, "production", DAY), {STAMP: 1200})
+        self.assertEqual(sum(form is not None and "password" in form for _, form, _, _ in client.requests), 2)
+
+    def test_persistent_denial_stops_after_one_session_renewal(self):
+        client = ScriptedClient(login_steps() + [(403, b"", "")] + login_steps() + [(403, b"", "")])
+        with self.assertRaisesRegex(TransportError, "No guest access"):
+            client.get_chart(SYSTEM, "production", DAY)
+        self.assertFalse(client.steps)
+
+    def test_rejected_credentials_are_submitted_only_once(self):
+        client = ScriptedClient(login_steps()[:2] + [(200, LOGIN_FORM, "")])
+        with self.assertRaisesRegex(TransportError, "login failed"):
+            client.get_chart(SYSTEM, "production", DAY)
+        self.assertEqual(sum(form is not None for _, form, _, _ in client.requests), 1)
+
+    def test_captcha_or_mfa_does_not_trigger_password_retries(self):
+        for body in [LOGIN_FORM.replace(b"</form>", b'<div data-sitekey="captcha"></div></form>'),
+                     b'<form method="post"><input name="otp"></form>']:
+            with self.subTest(body=body):
+                client = ScriptedClient([login_steps()[0], (200, body, "")])
+                with self.assertRaisesRegex(TransportError, "login failed"):
+                    client.get_chart(SYSTEM, "production", DAY)
+                self.assertFalse(any(form is not None for _, form, _, _ in client.requests))
+
+    def test_changed_or_untrusted_password_form_is_not_submitted(self):
+        bodies = [LOGIN_FORM.replace(b"/commonauth", b"https://evil.example/commonauth"),
+                  LOGIN_FORM.replace(b"/commonauth", b"/other"), LOGIN_FORM.replace(b'"post"', b'"get"'),
+                  LOGIN_FORM.replace(b"sessionDataKey", b"differentKey")]
+        for body in bodies:
+            with self.subTest(body=body):
+                client = ScriptedClient([login_steps()[0], (200, body, "")])
+                with self.assertRaises(TransportError):
+                    client.get_chart(SYSTEM, "production", DAY)
+                self.assertFalse(any(form is not None for _, form, _, _ in client.requests))
+
+    def test_untrusted_callback_is_never_sent_tokens(self):
+        for target in [b"https://evil.example/callback", b"https://login.fronius.com/commonauth",
+                       b"https://www.solarweb.com/other"]:
+            with self.subTest(target=target):
+                body = CALLBACK_FORM.replace(b"https://www.solarweb.com/Account/ExternalLoginCallback", target)
+                client = ScriptedClient(login_steps()[:2] + [(200, body, "")])
+                with self.assertRaises(TransportError):
+                    client.get_chart(SYSTEM, "production", DAY)
+                self.assertEqual(sum(form is not None for _, form, _, _ in client.requests), 1)
+
+    def test_guest_and_untrusted_redirects_are_blocked_before_network(self):
+        for target in [API_SERVER + "/Account/GuestLogOn?pvSystemId=x", API_SERVER + "/Account/DemoLogOn",
+                       "https://evil.example/", "http://www.solarweb.com/"]:
+            with self.subTest(target=target):
+                client = ScriptedClient([(302, b"", target)])
+                with self.assertRaisesRegex(TransportError, "Request blocked"):
+                    client.get_chart(SYSTEM, "production", DAY)
+                self.assertEqual(len(client.requests), 1)
+
+    def test_post_redirect_cannot_forward_credentials(self):
+        for status in [307, 308]:
+            with self.subTest(status=status):
+                client = ScriptedClient(login_steps()[:2] + [(status, b"", API_SERVER + "/")])
+                with self.assertRaisesRegex(TransportError, "sensitive POST"):
+                    client.get_chart(SYSTEM, "production", DAY)
+                self.assertEqual(len(client.requests), 3)
+
+
+class TransportTests(unittest.TestCase):
+    def test_url_allowlist(self):
+        for url in [API_SERVER + "/", LOGIN_SERVER + "/commonauth", API_SERVER + ":443/Chart/Chart"]:
+            validate_url(url)
+        for url in ["https://www.solarweb.com.evil.example/", "https://user:pass@www.solarweb.com/",
+                    API_SERVER + ":444/", "http://www.solarweb.com/", API_SERVER + "/%47uestLogOn",
+                    API_SERVER + "/DemoLogOn", API_SERVER + "/bad\npath", API_SERVER + "\\@evil.example/"]:
+            with self.subTest(url=url), self.assertRaises(TransportError):
+                validate_url(url)
+
+    def test_transport_redacts_network_errors_and_response_repr(self):
+        client = SolarWebClient("account@example.test", PASSWORD)
+        client.opener = Mock()
+        client.opener.open.side_effect = URLError("synthetic-token " + PASSWORD)
+        with self.assertRaises(TransportError) as raised:
+            client._request(API_SERVER + "/")
+        self.assertNotIn(PASSWORD, str(raised.exception))
+        self.assertNotIn("synthetic-token", str(raised.exception))
+        self.assertEqual(repr(Response(200, API_SERVER + "/?token=synthetic-token", b"secret", "secret")), "Response(status=200)")
+
+    def test_transport_keeps_sso_queries_out_of_referer_and_encodes_password(self):
+        client = SolarWebClient("account@example.test", PASSWORD)
+        response = BytesIO(b"success")
+        response.code, response.headers = 200, {}
+        client.opener = Mock()
+        client.opener.open.return_value = response
+        client._request(LOGIN_SERVER + "/commonauth", {"password": PASSWORD}, LOGIN_SERVER + "/?state=synthetic-token")
+        request = client.opener.open.call_args.args[0]
+        self.assertEqual(request.get_header("Referer"), LOGIN_SERVER + "/")
+        self.assertEqual(request.get_header("Origin"), LOGIN_SERVER)
+        self.assertEqual(parse_qs(request.data.decode())["password"], [PASSWORD])
+
+    def test_responses_are_bounded(self):
+        client = SolarWebClient("account@example.test", PASSWORD)
+        response = BytesIO(b"x" * (client.MAX_RESPONSE_BYTES + 1))
+        response.code, response.headers = 200, {}
+        client.opener = Mock()
+        client.opener.open.return_value = response
+        with self.assertRaisesRegex(TransportError, "exceeds 8 MiB"):
+            client._request(API_SERVER + "/")
+
+
+class ChartTests(unittest.TestCase):
+    def test_series_uses_id_and_skips_null_instead_of_zeroing(self):
+        self.assertEqual(decode_chart(chart(), "FromGen"), {STAMP: 1200})
+
+    def test_bad_units_missing_series_and_invalid_points_stop_import(self):
+        bodies = [b"not JSON", b'{"settings":{"series":[]}}', chart("Other"), chart(suffix="kW"),
+                  chart(points=[[STAMP + 1, 1200]]), chart(points=[[STAMP, True]]),
+                  chart(points=[[STAMP, -1]]), chart(points=[[STAMP, float("nan")]]),
+                  chart(points=[[STAMP, 1], [STAMP, 2]]), chart(points=[[True, 1]]),
+                  chart(points=[[STAMP, "1200"]]), chart(points=[[STAMP]])]
+        for body in bodies:
+            with self.subTest(body=body), self.assertRaises(TransportError):
+                decode_chart(body, "FromGen")
+
+
+if __name__ == "__main__":
+    unittest.main()