mirrors/pyhOn
1
0
Fork 0
mirror of https://github.com/Andre0512/pyhOn.git synced 2024-12-22 18:55:32 +00:00
Code Issues Packages Projects Releases Wiki Activity

Fix problems with refresh token

Browse source
This commit is contained in:
Andre Basche 2024-02-11 02:02:47 +01:00
parent 767cbe35de
commit c4e8038652
4 changed files with 9 additions and 8 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
pyhon/command_loader.py
View file

@ -109,7 +109,7 @@ class HonCommandLoader:
categories: Optional[Dict[str, "HonCommand"]] = None, categories: Optional[Dict[str, "HonCommand"]] = None,
category_name: str = "", category_name: str = "",
) -> Optional[HonCommand]: ) -> Optional[HonCommand]:
"""Try to crate HonCommand object""" """Try to create HonCommand object"""
if not isinstance(data, dict): if not isinstance(data, dict):
self._additional_data[command_name] = data self._additional_data[command_name] = data
return None return None

2
pyhon/connection/api.py
View file

@ -75,7 +75,7 @@ class HonAPI:
self._hon_handler = await HonConnectionHandler( self._hon_handler = await HonConnectionHandler(
self._email, self._email,
self._password, self._password,
self._session, session=self._session,
mobile_id=self._mobile_id, mobile_id=self._mobile_id,
).create() ).create()
return self return self

8
pyhon/connection/auth.py
View file

@ -48,7 +48,6 @@ class HonAuth:
email: str, email: str,
password: str, password: str,
device: HonDevice, device: HonDevice,
refresh_token: str = "",
) -> None: ) -> None:
self._session = session self._session = session
self._request = HonAuthConnectionHandler(session) self._request = HonAuthConnectionHandler(session)
@ -58,7 +57,6 @@ class HonAuth:
self._device = device self._device = device
self._expires: datetime = datetime.utcnow() self._expires: datetime = datetime.utcnow()
self._auth = HonAuthData() self._auth = HonAuthData()
self._auth.refresh_token = refresh_token
@property @property
def cognito_token(self) -> str: def cognito_token(self) -> str:
@ -201,7 +199,7 @@ class HonAuth:
if access_token := re.findall("access_token=(.*?)&", text): if access_token := re.findall("access_token=(.*?)&", text):
self._auth.access_token = access_token[0] self._auth.access_token = access_token[0]
if refresh_token := re.findall("refresh_token=(.*?)&", text): if refresh_token := re.findall("refresh_token=(.*?)&", text):
self._auth.refresh_token = refresh_token[0] self._auth.refresh_token = parse.unquote(refresh_token[0])
if id_token := re.findall("id_token=(.*?)&", text): if id_token := re.findall("id_token=(.*?)&", text):
self._auth.id_token = id_token[0] self._auth.id_token = id_token[0]
return bool(access_token and refresh_token and id_token) return bool(access_token and refresh_token and id_token)
@ -266,7 +264,9 @@ class HonAuth:
except exceptions.HonNoAuthenticationNeeded: except exceptions.HonNoAuthenticationNeeded:
return return
async def refresh(self) -> bool: async def refresh(self, refresh_token="") -> bool:
if refresh_token:
self._auth.refresh_token = refresh_token
params = { params = {
"client_id": const.CLIENT_ID, "client_id": const.CLIENT_ID,
"refresh_token": self._auth.refresh_token, "refresh_token": self._auth.refresh_token,

5
pyhon/connection/handler/hon.py
View file

@ -22,9 +22,9 @@ class HonConnectionHandler(ConnectionHandler):
self, self,
email: str, email: str,
password: str, password: str,
session: Optional[aiohttp.ClientSession] = None,
mobile_id: str = "", mobile_id: str = "",
refresh_token: str = "", refresh_token: str = "",
session: Optional[aiohttp.ClientSession] = None,
) -> None: ) -> None:
super().__init__(session=session) super().__init__(session=session)
self._device: HonDevice = HonDevice(mobile_id) self._device: HonDevice = HonDevice(mobile_id)
@ -54,11 +54,12 @@ class HonConnectionHandler(ConnectionHandler):
self._email, self._email,
self._password, self._password,
self._device, self._device,
refresh_token=self._refresh_token,
) )
return self return self
async def _check_headers(self, headers: Dict[str, str]) -> Dict[str, str]: async def _check_headers(self, headers: Dict[str, str]) -> Dict[str, str]:
if self._refresh_token:
await self.auth.refresh(self._refresh_token)
if not (self.auth.cognito_token and self.auth.id_token): if not (self.auth.cognito_token and self.auth.id_token):
await self.auth.authenticate() await self.auth.authenticate()
headers["cognito-token"] = self.auth.cognito_token headers["cognito-token"] = self.auth.cognito_token