pyhOn/pyhon/connection/auth.py

269 lines
10 KiB
Python
Raw Normal View History

2023-02-13 01:41:38 +01:00
import json
import logging
import re
import secrets
import urllib
from datetime import datetime, timedelta
2023-04-10 06:42:40 +02:00
from pprint import pformat
2023-04-13 23:25:49 +02:00
from typing import List, Tuple
2023-02-13 01:41:38 +01:00
from urllib import parse
2023-04-12 02:09:41 +02:00
from urllib.parse import quote
2023-02-13 01:41:38 +01:00
2023-03-17 01:56:04 +01:00
from yarl import URL
2023-02-13 01:41:38 +01:00
2023-04-12 19:14:14 +02:00
from pyhon import const, exceptions
2023-02-13 01:41:38 +01:00
2023-04-10 20:31:55 +02:00
_LOGGER = logging.getLogger(__name__)
2023-02-13 01:41:38 +01:00
class HonAuth:
_TOKEN_EXPIRES_AFTER_HOURS = 8
_TOKEN_EXPIRE_WARNING_HOURS = 7
2023-04-09 18:13:50 +02:00
def __init__(self, session, email, password, device) -> None:
self._session = session
self._email = email
self._password = password
2023-03-17 01:56:04 +01:00
self._access_token = ""
self._refresh_token = ""
2023-02-13 01:41:38 +01:00
self._cognito_token = ""
self._id_token = ""
2023-04-09 18:13:50 +02:00
self._device = device
2023-04-13 23:25:49 +02:00
self._called_urls: List[Tuple[int, str]] = []
self._expires: datetime = datetime.utcnow()
2023-02-13 01:41:38 +01:00
@property
def cognito_token(self):
return self._cognito_token
@property
def id_token(self):
return self._id_token
2023-03-17 01:56:04 +01:00
@property
def access_token(self):
return self._access_token
@property
def refresh_token(self):
return self._refresh_token
def _check_token_expiration(self, hours):
return datetime.utcnow() >= self._expires + timedelta(hours=hours)
@property
def token_is_expired(self) -> bool:
return self._check_token_expiration(self._TOKEN_EXPIRES_AFTER_HOURS)
@property
def token_expires_soon(self) -> bool:
return self._check_token_expiration(self._TOKEN_EXPIRE_WARNING_HOURS)
async def _error_logger(self, response, fail=True):
result = "hOn Authentication Error\n"
for i, (status, url) in enumerate(self._called_urls):
result += f" {i + 1: 2d} {status} - {url}\n"
result += f"ERROR - {response.status} - {response.request_info.url}\n"
result += f"{15 * '='} Response {15 * '='}\n{await response.text()}\n{40 * '='}"
_LOGGER.error(result)
if fail:
2023-04-12 19:14:14 +02:00
raise exceptions.HonAuthenticationError("Can't login")
2023-04-09 18:13:50 +02:00
async def _load_login(self):
2023-03-17 01:56:04 +01:00
nonce = secrets.token_hex(16)
nonce = f"{nonce[:8]}-{nonce[8:12]}-{nonce[12:16]}-{nonce[16:20]}-{nonce[20:]}"
params = {
"response_type": "token+id_token",
"client_id": const.CLIENT_ID,
2023-04-09 20:55:36 +02:00
"redirect_uri": urllib.parse.quote(
f"{const.APP}://mobilesdk/detect/oauth/done"
),
2023-03-17 01:56:04 +01:00
"display": "touch",
"scope": "api openid refresh_token web",
2023-04-09 20:55:36 +02:00
"nonce": nonce,
2023-03-17 01:56:04 +01:00
}
params = "&".join([f"{k}={v}" for k, v in params.items()])
2023-04-09 20:55:36 +02:00
async with self._session.get(
f"{const.AUTH_API}/services/oauth2/authorize/expid_Login?{params}"
2023-04-10 20:31:55 +02:00
) as response:
self._called_urls.append((response.status, response.request_info.url))
2023-04-12 19:14:14 +02:00
text = await response.text()
self._expires = datetime.utcnow()
2023-04-12 19:14:14 +02:00
if not (login_url := re.findall("url = '(.+?)'", text)):
if "oauth/done#access_token=" in text:
self._parse_token_data(text)
raise exceptions.HonNoAuthenticationNeeded()
await self._error_logger(response)
2023-03-17 01:56:04 +01:00
return False
2023-04-09 18:13:50 +02:00
async with self._session.get(login_url[0], allow_redirects=False) as redirect1:
self._called_urls.append((redirect1.status, redirect1.request_info.url))
2023-03-17 01:56:04 +01:00
if not (url := redirect1.headers.get("Location")):
await self._error_logger(redirect1)
2023-03-17 01:56:04 +01:00
return False
2023-04-09 18:13:50 +02:00
async with self._session.get(url, allow_redirects=False) as redirect2:
self._called_urls.append((redirect2.status, redirect2.request_info.url))
2023-04-09 20:55:36 +02:00
if not (
url := redirect2.headers.get("Location")
+ "&System=IoT_Mobile_App&RegistrationSubChannel=hOn"
):
await self._error_logger(redirect2)
2023-03-17 01:56:04 +01:00
return False
2023-04-12 01:07:03 +02:00
async with self._session.get(
URL(url, encoded=True), headers={"user-agent": const.USER_AGENT}
) as login_screen:
self._called_urls.append(
(login_screen.status, login_screen.request_info.url)
)
2023-04-09 20:55:36 +02:00
if context := re.findall(
'"fwuid":"(.*?)","loaded":(\\{.*?})', await login_screen.text()
):
2023-03-17 01:56:04 +01:00
fw_uid, loaded_str = context[0]
loaded = json.loads(loaded_str)
2023-04-09 20:55:36 +02:00
login_url = login_url[0].replace(
"/".join(const.AUTH_API.split("/")[:-1]), ""
)
2023-03-17 01:56:04 +01:00
return fw_uid, loaded, login_url
await self._error_logger(login_screen)
2023-03-17 01:56:04 +01:00
return False
2023-04-09 18:13:50 +02:00
async def _login(self, fw_uid, loaded, login_url):
2023-02-13 01:41:38 +01:00
data = {
"message": {
"actions": [
{
"id": "79;a",
"descriptor": "apex://LightningLoginCustomController/ACTION$login",
"callingDescriptor": "markup://c:loginForm",
"params": {
2023-04-12 02:09:41 +02:00
"username": quote(self._email),
"password": quote(self._password),
2023-04-09 20:55:36 +02:00
"startUrl": parse.unquote(
login_url.split("startURL=")[-1]
).split("%3D")[0],
},
2023-02-13 01:41:38 +01:00
}
]
},
"aura.context": {
"mode": "PROD",
2023-03-17 01:56:04 +01:00
"fwuid": fw_uid,
2023-02-13 01:41:38 +01:00
"app": "siteforce:loginApp2",
2023-03-17 01:56:04 +01:00
"loaded": loaded,
2023-02-13 01:41:38 +01:00
"dn": [],
"globals": {},
2023-04-09 20:55:36 +02:00
"uad": False,
},
2023-03-17 01:56:04 +01:00
"aura.pageURI": login_url,
2023-04-09 20:55:36 +02:00
"aura.token": None,
}
2023-02-13 01:41:38 +01:00
params = {"r": 3, "other.LightningLoginCustom.login": 1}
2023-04-09 18:13:50 +02:00
async with self._session.post(
2023-04-09 20:55:36 +02:00
const.AUTH_API + "/s/sfsites/aura",
headers={"Content-Type": "application/x-www-form-urlencoded"},
data="&".join(f"{k}={json.dumps(v)}" for k, v in data.items()),
params=params,
2023-02-13 01:41:38 +01:00
) as response:
self._called_urls.append((response.status, response.request_info.url))
2023-03-17 01:56:04 +01:00
if response.status == 200:
try:
2023-04-10 06:42:40 +02:00
data = await response.json()
return data["events"][0]["attributes"]["values"]["url"]
2023-03-17 01:56:04 +01:00
except json.JSONDecodeError:
pass
2023-04-10 06:42:40 +02:00
except KeyError:
_LOGGER.error(
"Can't get login url - %s", pformat(await response.json())
)
await self._error_logger(response)
2023-03-17 01:56:04 +01:00
return ""
2023-02-13 01:41:38 +01:00
2023-04-12 19:14:14 +02:00
def _parse_token_data(self, text):
if access_token := re.findall("access_token=(.*?)&", text):
self._access_token = access_token[0]
if refresh_token := re.findall("refresh_token=(.*?)&", text):
self._refresh_token = refresh_token[0]
if id_token := re.findall("id_token=(.*?)&", text):
self._id_token = id_token[0]
2023-04-09 18:13:50 +02:00
async def _get_token(self, url):
2023-04-10 20:31:55 +02:00
async with self._session.get(url) as response:
self._called_urls.append((response.status, response.request_info.url))
2023-04-10 20:31:55 +02:00
if response.status != 200:
await self._error_logger(response)
2023-03-17 01:56:04 +01:00
return False
2023-04-10 20:31:55 +02:00
url = re.findall("href\\s*=\\s*[\"'](.+?)[\"']", await response.text())
if not url:
await self._error_logger(response)
return False
2023-04-10 20:31:55 +02:00
if "ProgressiveLogin" in url[0]:
async with self._session.get(url[0]) as response:
self._called_urls.append((response.status, response.request_info.url))
2023-04-10 20:31:55 +02:00
if response.status != 200:
await self._error_logger(response)
2023-04-10 20:31:55 +02:00
return False
url = re.findall("href\\s*=\\s*[\"'](.*?)[\"']", await response.text())
url = "/".join(const.AUTH_API.split("/")[:-1]) + url[0]
async with self._session.get(url) as response:
self._called_urls.append((response.status, response.request_info.url))
2023-04-10 20:31:55 +02:00
if response.status != 200:
await self._error_logger(response)
2023-02-13 01:41:38 +01:00
return False
2023-04-12 19:14:14 +02:00
self._parse_token_data(await response.text())
2023-02-13 01:41:38 +01:00
return True
2023-04-11 17:09:02 +02:00
async def _api_auth(self):
2023-04-09 18:43:57 +02:00
post_headers = {"id-token": self._id_token}
2023-04-09 18:13:50 +02:00
data = self._device.get()
2023-04-09 20:55:36 +02:00
async with self._session.post(
f"{const.API_URL}/auth/v1/login", headers=post_headers, json=data
2023-04-10 20:31:55 +02:00
) as response:
self._called_urls.append((response.status, response.request_info.url))
2023-04-09 18:13:50 +02:00
try:
2023-04-10 20:31:55 +02:00
json_data = await response.json()
2023-04-09 18:13:50 +02:00
except json.JSONDecodeError:
await self._error_logger(response)
2023-04-09 18:13:50 +02:00
return False
self._cognito_token = json_data["cognitoUser"]["Token"]
2023-02-13 01:41:38 +01:00
return True
2023-04-09 18:43:57 +02:00
2023-04-12 19:14:14 +02:00
async def authenticate(self):
self.clear()
try:
if not (login_site := await self._load_login()):
raise exceptions.HonAuthenticationError("Can't open login page")
if not (url := await self._login(*login_site)):
raise exceptions.HonAuthenticationError("Can't login")
if not await self._get_token(url):
raise exceptions.HonAuthenticationError("Can't get token")
if not await self._api_auth():
raise exceptions.HonAuthenticationError("Can't get api token")
except exceptions.HonNoAuthenticationNeeded:
return
2023-04-09 18:43:57 +02:00
async def refresh(self):
params = {
"client_id": const.CLIENT_ID,
"refresh_token": self._refresh_token,
2023-04-09 20:55:36 +02:00
"grant_type": "refresh_token",
2023-04-09 18:43:57 +02:00
}
2023-04-09 20:55:36 +02:00
async with self._session.post(
f"{const.AUTH_API}/services/oauth2/token", params=params
2023-04-10 20:31:55 +02:00
) as response:
self._called_urls.append((response.status, response.request_info.url))
2023-04-10 20:31:55 +02:00
if response.status >= 400:
await self._error_logger(response, fail=False)
2023-04-09 18:43:57 +02:00
return False
2023-04-10 20:31:55 +02:00
data = await response.json()
self._expires = datetime.utcnow()
2023-04-10 06:42:40 +02:00
self._id_token = data["id_token"]
self._access_token = data["access_token"]
2023-04-11 17:09:02 +02:00
return await self._api_auth()
2023-04-12 19:14:14 +02:00
def clear(self):
self._session.cookie_jar.clear_domain(const.AUTH_API.split("/")[-2])
self._called_urls = []
2023-04-12 19:14:14 +02:00
self._cognito_token = ""
self._id_token = ""
self._access_token = ""
self._refresh_token = ""