Fix User Agent issue again

pymyq/__version__.py

@@ -1,2 +1,2 @@
 """Define a version constant."""
-__version__ = '3.0.3'
+__version__ = "3.0.4"

pymyq/api.py

@@ -5,6 +5,8 @@
 from datetime import datetime, timedelta
 from typing import Dict, Optional, Union, Tuple
 from urllib.parse import urlsplit, parse_qs
+from random import choices
+import string
 
 from aiohttp import ClientSession, ClientResponse
 from aiohttp.client_exceptions import ClientError, ClientResponseError
@@ -39,11 +41,17 @@ class API:  # pylint: disable=too-many-instance-attributes
     """Define a class for interacting with the MyQ iOS App API."""
 
     def __init__(
-        self, username: str, password: str, websession: ClientSession = None
+        self,
+        username: str,
+        password: str,
+        websession: ClientSession = None,
+        useragent: Optional[str] = None,
     ) -> None:
         """Initialize."""
         self.__credentials = {"username": username, "password": password}
-        self._myqrequests = MyQRequest(websession or ClientSession())
+        self._myqrequests = MyQRequest(
+            websession or ClientSession(), useragent=useragent
+        )
         self._authentication_task = None  # type:Optional[asyncio.Task]
         self._codeverifier = None  # type: Optional[str]
         self._invalid_credentials = False  # type: bool
@@ -381,7 +389,6 @@ async def _oauth_authenticate(self) -> Tuple[str, int]:
                 websession=session,
                 headers={
                     "Cookie": resp.cookies.output(attrs=[]),
-                    "User-Agent": "null",
                 },
                 allow_redirects=False,
                 login_request=True,
@@ -398,7 +405,6 @@ async def _oauth_authenticate(self) -> Tuple[str, int]:
                 websession=session,
                 headers={
                     "Content-Type": "application/x-www-form-urlencoded",
-                    "User-Agent": "null",
                 },
                 data={
                     "client_id": OAUTH_CLIENT_ID,
@@ -641,8 +647,48 @@ async def update_device_info(self, for_account: str = None) -> None:
 async def login(username: str, password: str, websession: ClientSession = None) -> API:
     """Log in to the API."""
 
+    # Retrieve user agent from GitHub if not provided for login.
+    _LOGGER.debug("No user agent provided, trying to retrieve from GitHub.")
+    url = f"https://raw.githubusercontent.com/arraylabs/pymyq/master/.USER_AGENT"
+
+    try:
+        async with ClientSession() as session:
+            async with session.get(url) as resp:
+                useragent = await resp.text()
+                resp.raise_for_status()
+                _LOGGER.debug(f"Retrieved user agent {useragent} from GitHub.")
+
+    except ClientError as exc:
+        # Default user agent to random string with length of 5 if failure to retrieve it from GitHub.
+        useragent = "#RANDOM:5"
+        _LOGGER.warning(
+            f"Failed retrieving user agent from GitHub, will use randomized user agent "
+            f"instead: {str(exc)}"
+        )
+
+    # Check if value for useragent is to create a random user agent.
+    useragent_list = useragent.split(":")
+    if useragent_list[0] == "#RANDOM":
+        # Create a random string, check if length is provided for the random string, if not then default is 5.
+        try:
+            randomlength = int(useragent_list[1]) if len(useragent_list) == 2 else 5
+        except ValueError:
+            _LOGGER.debug(
+                f"Random length value {useragent_list[1]} in user agent {useragent} is not an integer. "
+                f"Setting to 5 instead."
+            )
+            randomlength = 5
+
+        # Create the random user agent.
+        useragent = "".join(
+            choices(string.ascii_letters + string.digits, k=randomlength)
+        )
+        _LOGGER.debug(f"User agent set to randomized value: {useragent}.")
+
     # Set the user agent in the headers.
-    api = API(username=username, password=password, websession=websession)
+    api = API(
+        username=username, password=password, websession=websession, useragent=useragent
+    )
     _LOGGER.debug("Performing initial authentication into MyQ")
     try:
         await api.authenticate(wait=True)

pymyq/request.py

@@ -19,11 +19,12 @@
 class MyQRequest:  # pylint: disable=too-many-instance-attributes
     """Define a class to handle requests to MyQ"""
 
-    def __init__(self, websession: ClientSession = None) -> None:
+    def __init__(self, websession: ClientSession = None, useragent: str = None) -> None:
         self._websession = websession or ClientSession()
+        self._useragent = useragent
 
-    @staticmethod
     async def _send_request(
+        self,
         method: str,
         url: str,
         websession: ClientSession,
@@ -38,16 +39,24 @@ async def _send_request(
         resp_exc = None
         last_status = ""
         last_error = ""
+
+        if self._useragent is not None:
+            headers.update({"User-Agent": self._useragent})
+
         while attempt < DEFAULT_REQUEST_RETRIES:
             if attempt != 0:
                 wait_for = min(2 ** attempt, 5)
-                _LOGGER.debug(f'Request failed with "{last_status} {last_error}" '
-                              f'(attempt #{attempt}/{DEFAULT_REQUEST_RETRIES})"; trying again in {wait_for} seconds')
+                _LOGGER.debug(
+                    f'Request failed with "{last_status} {last_error}" '
+                    f'(attempt #{attempt}/{DEFAULT_REQUEST_RETRIES})"; trying again in {wait_for} seconds'
+                )
                 await asyncio.sleep(wait_for)
 
             attempt += 1
             try:
-                _LOGGER.debug(f"Sending myq api request {url} and headers {headers} with connection pooling")
+                _LOGGER.debug(
+                    f"Sending myq api request {url} and headers {headers} with connection pooling"
+                )
                 resp = await websession.request(
                     method,
                     url,