Updated SMUD login (#77)

* first try at adding smud

* login sucessful unable to parse opower token

* not yet working, token is in history

* proof of concept getsokta token
this works as far as getting the token

* SMUD, tested.

* @v-yarotsky code review changes

* Fix redirects logging

Was previously showing the requested URL and the redirect and stating there were two redirects (where there was two requests)

* Ignore `.env.secret`

* Add pytest to optional dependencies

* use `len()`

* If we already have a cookie, return early if it is valid.

* 0.4.0

* Update __init__.py

src/opower/__init__.py:5:5: F401 [*] `.opower.DEBUG_LOG_RESPONSE` imported but unused

* Update test_smud.py

* Set types for mypy

* Use Optional[str]

* Update README.md

* Update smud.py

* Update `-vv` changes from #76

* Delete cookies if they don't work.

* Update smud.py

* Update smud.py

---------

Co-authored-by: andylittle <[email protected]>
Co-authored-by: tronikos <[email protected]>

src/opower/utilities/smud.py

@@ -44,13 +44,46 @@ def __init__(self) -> None:
         """Initialize."""
         super().__init__()
         self.verification_token: Optional[str] = None
+        self.ocis_req_sp: Optional[str] = None
+        self.relay_state: Optional[str] = None
 
     def handle_starttag(self, tag: str, attrs: list[tuple[str, Optional[str]]]) -> None:
         """Try to extract the verification token from the login input."""
         if tag == "input" and ("name", "__RequestVerificationToken") in attrs:
             _, token = next(filter(lambda attr: attr[0] == "value", attrs))
-            _LOGGER.debug("SMUD self verify token: %s", token)
+            if token is None:
+                return
+            _LOGGER.debug(
+                "SMUD self verify token: %s...%s (%d characters)",
+                token[0:5],
+                token[-5:],
+                len(token),
+            )
             self.verification_token = token
+        """Try to extract the OCIS_REQ_SP input value from the identity.oraclecloud.com sso HTML"""
+        if tag == "input" and ("name", "OCIS_REQ_SP") in attrs:
+            _, token = next(filter(lambda attr: attr[0] == "value", attrs))
+            if token is None:
+                return
+            _LOGGER.debug(
+                "OCIS_REQ_SP self verify token: %s...%s (%d characters)",
+                token[0:5],
+                token[-5:],
+                len(token),
+            )
+            self.ocis_req_sp = token
+        """Try to extract the RelayState input value from the smud.okta.com HTML"""
+        if tag == "input" and ("name", "RelayState") in attrs:
+            _, token = next(filter(lambda attr: attr[0] == "value", attrs))
+            if token is None:
+                return
+            _LOGGER.debug(
+                "RelayState value: %s...%s (%d characters)",
+                token[0:5],
+                token[-5:],
+                len(token),
+            )
+            self.relay_state = token
 
 
 class SMUDOktaResponseSamlResponseValueParser(HTMLParser):
@@ -104,6 +137,8 @@ async def async_login(
                 ):
                     return
             except ClientResponseError:
+                _LOGGER.debug("Failed to login to SMUD with existing cookies")
+                session.cookie_jar.clear()
                 pass
 
         smud_login_page_url = "https://myaccount.smud.org/"
@@ -188,19 +223,19 @@ async def async_login(
         )
 
         # This step is done in the web browser but doesn't seem to matter here.
-        #
-        # smud_ssotransition_url = "https://myaccount.smud.org/signin/ssotransition"
-        #
-        # _LOGGER.debug("Fetching SMUD ssotransition page: %s", smud_ssotransition_url)
-        # smud_ssotransition_response = await session.get(
-        #     smud_ssotransition_url,
-        #     headers={"User-Agent": USER_AGENT},
-        #     raise_for_status=True,
-        # )
-        #
-        # SMUD.print_redirects_cookies_response(smud_ssotransition_response, session)
-
-        opower_sso_url = "https://sso.opower.com/sp/ACS.saml2"
+        smud_ssotransition_url = "https://myaccount.smud.org/signin/ssotransition"
+
+        _LOGGER.debug("Fetching SMUD ssotransition page: %s", smud_ssotransition_url)
+        smud_ssotransition_response = await session.get(
+            smud_ssotransition_url,
+            headers={"User-Agent": USER_AGENT},
+            raise_for_status=True,
+        )
+
+        await SMUD.log_response(smud_ssotransition_response, session)
+
+        # This is the action of the #appForm form in the smud_okta_response HTML.
+        opower_sso_url = "https://idcs-8d184356671642c58ea38b42e6420ed2.identity.oraclecloud.com/fed/v1/sp/sso"
 
         _LOGGER.debug("POSTing opower sso page with SAMLResponse: %s", opower_sso_url)
 
@@ -212,10 +247,77 @@ async def async_login(
             },
             headers={"User-Agent": USER_AGENT},
             raise_for_status=True,
+            allow_redirects=True,
         )
-
         await SMUD.log_response(opower_sso_response, session)
 
+        login_parser.feed(await opower_sso_response.text())
+        ocis_req_sp = login_parser.ocis_req_sp
+
+        identity_oraclecloud_login_url = "https://idcs-8d184356671642c58ea38b42e6420ed2.identity.oraclecloud.com/sso/v1/user/login"
+
+        _LOGGER.debug(
+            "POSTing opower sso login page with OCIS_REQ_SP: %s",
+            identity_oraclecloud_login_url,
+        )
+
+        identity_oraclecloud_login_response = await session.post(
+            identity_oraclecloud_login_url,
+            data={
+                "OCIS_REQ_SP": ocis_req_sp,
+            },
+            headers={"User-Agent": USER_AGENT},
+            raise_for_status=True,
+            allow_redirects=True,
+            max_redirects=10,
+        )
+
+        await SMUD.log_response(identity_oraclecloud_login_response, session)
+
+        okta_saml_request_url = identity_oraclecloud_login_response.real_url
+
+        _LOGGER.debug(
+            "Fetching okta saml page with SAMLRequest, RelayState: %s",
+            okta_saml_request_url,
+        )
+
+        okta_saml_request_response = await session.get(
+            okta_saml_request_url,
+            headers={"User-Agent": USER_AGENT},
+            raise_for_status=True,
+        )
+
+        await SMUD.log_response(okta_saml_request_response, session)
+
+        parser2 = SMUDOktaResponseSamlResponseValueParser()
+        parser2.feed(await okta_saml_request_response.text())
+        saml_response = parser2.saml_response
+        assert saml_response
+
+        login_parser.feed(await okta_saml_request_response.text())
+        relay_state = login_parser.relay_state
+
+        opower_sso_acs_url = "https://sso.opower.com/sp/ACS.saml2"
+
+        _LOGGER.debug(
+            "POSTing opower ACS sso page with SAMLResponse: %s", opower_sso_acs_url
+        )
+
+        opower_sso_acs_response = await session.post(
+            opower_sso_acs_url,
+            data={
+                "SAMLResponse": saml_response,
+                "RelayState": relay_state,
+            },
+            headers={"User-Agent": USER_AGENT},
+            raise_for_status=True,
+            allow_redirects=True,
+        )
+
+        await SMUD.log_response(opower_sso_acs_response, session)
+
+        _LOGGER.debug("End of SMUD login process")
+
         return
 
     @classmethod