Split login logic

README.md

@@ -38,15 +38,19 @@ import asyncio
 import time
 
 from pyoverkiz.const import SUPPORTED_SERVERS
-from pyoverkiz.client import OverkizClient
+from aiohttp import ClientSession
 
 USERNAME = ""
 PASSWORD = ""
 
+
 async def main() -> None:
-    async with OverkizClient(USERNAME, PASSWORD, server=SUPPORTED_SERVERS["somfy_europe"]) as client:
+
+    session = ClientSession()
+    client = SUPPORTED_SERVERS["somfy_europe"](session)
+    async with client as client:
         try:
-            await client.login()
+            await client.login(USERNAME, PASSWORD)
         except Exception as exception:  # pylint: disable=broad-except
             print(exception)
             return
@@ -63,7 +67,9 @@ async def main() -> None:
 
             time.sleep(2)
 
+
 asyncio.run(main())
+
 ```
 
 ## Development

mypy.ini

@@ -3,18 +3,22 @@ python_version = 3.8
 show_error_codes = true
 follow_imports = silent
 ignore_missing_imports = true
+local_partial_types = true
 strict_equality = true
+no_implicit_optional = true
 warn_incomplete_stub = true
 warn_redundant_casts = true
 warn_unused_configs = true
 warn_unused_ignores = true
+enable_error_code = ignore-without-code, redundant-self, truthy-iterable
+disable_error_code = annotation-unchecked
+strict_concatenate = false
 check_untyped_defs = true
 disallow_incomplete_defs = true
 disallow_subclassing_any = true
 disallow_untyped_calls = true
-#disallow_untyped_decorators = true
+disallow_untyped_decorators = true
 disallow_untyped_defs = true
-no_implicit_optional = true
 warn_return_any = true
 warn_unreachable = true
 

pyoverkiz/clients/atlantic_cozytouch.py

@@ -0,0 +1,64 @@
+from __future__ import annotations
+
+from aiohttp import FormData
+
+from pyoverkiz.clients.overkiz import OverkizClient
+from pyoverkiz.exceptions import (
+    CozyTouchBadCredentialsException,
+    CozyTouchServiceException,
+)
+
+COZYTOUCH_ATLANTIC_API = "https://apis.groupe-atlantic.com"
+COZYTOUCH_CLIENT_ID = (
+    "Q3RfMUpWeVRtSUxYOEllZkE3YVVOQmpGblpVYToyRWNORHpfZHkzNDJVSnFvMlo3cFNKTnZVdjBh"
+)
+
+
+class AtlanticCozytouchClient(OverkizClient):
+    async def _login(self, username: str, password: str) -> bool:
+        """
+        Authenticate and create an API session allowing access to the other operations.
+        Caller must provide one of [userId+userPassword, userId+ssoToken, accessToken, jwt]
+        """
+
+        async with self.session.post(
+            COZYTOUCH_ATLANTIC_API + "/token",
+            data=FormData(
+                {
+                    "grant_type": "password",
+                    "username": "GA-PRIVATEPERSON/" + username,
+                    "password": password,
+                }
+            ),
+            headers={
+                "Authorization": f"Basic {COZYTOUCH_CLIENT_ID}",
+                "Content-Type": "application/x-www-form-urlencoded",
+            },
+        ) as response:
+            token = await response.json()
+
+            # {'error': 'invalid_grant',
+            # 'error_description': 'Provided Authorization Grant is invalid.'}
+            if "error" in token and token["error"] == "invalid_grant":
+                raise CozyTouchBadCredentialsException(token["error_description"])
+
+            if "token_type" not in token:
+                raise CozyTouchServiceException("No CozyTouch token provided.")
+
+        # Request JWT
+        async with self.session.get(
+            COZYTOUCH_ATLANTIC_API + "/magellan/accounts/jwt",
+            headers={"Authorization": f"Bearer {token['access_token']}"},
+        ) as response:
+            jwt = await response.text()
+
+            if not jwt:
+                raise CozyTouchServiceException("No JWT token provided.")
+
+            jwt = jwt.strip('"')  # Remove surrounding quotes
+
+        payload = {"jwt": jwt}
+
+        post_response = await self.post("login", data=payload)
+
+        return "success" in post_response

pyoverkiz/clients/default.py

@@ -0,0 +1,12 @@
+from pyoverkiz.clients.overkiz import OverkizClient
+
+
+class DefaultClient(OverkizClient):
+    async def _login(self, username: str, password: str) -> bool:
+        """
+        Authenticate and create an API session allowing access to the other operations.
+        Caller must provide one of [userId+userPassword, userId+ssoToken, accessToken, jwt]
+        """
+        payload = {"userId": username, "userPassword": password}
+        response = await self.post("login", data=payload)
+        return "success" in response

pyoverkiz/clients/nexity.py

@@ -0,0 +1,67 @@
+from __future__ import annotations
+
+import asyncio
+from typing import cast
+
+import boto3
+from botocore.config import Config
+from warrant_lite import WarrantLite
+
+from pyoverkiz.clients.overkiz import OverkizClient
+from pyoverkiz.exceptions import NexityBadCredentialsException, NexityServiceException
+
+NEXITY_API = "https://api.egn.prd.aws-nexity.fr"
+NEXITY_COGNITO_CLIENT_ID = "3mca95jd5ase5lfde65rerovok"
+NEXITY_COGNITO_USER_POOL = "eu-west-1_wj277ucoI"
+NEXITY_COGNITO_REGION = "eu-west-1"
+
+
+class NexityClient(OverkizClient):
+    async def _login(self, username: str, password: str) -> bool:
+        """
+        Authenticate and create an API session allowing access to the other operations.
+        Caller must provide one of [userId+userPassword, userId+ssoToken, accessToken, jwt]
+        """
+
+        loop = asyncio.get_event_loop()
+
+        def _get_client() -> boto3.session.Session.client:
+            return boto3.client(
+                "cognito-idp", config=Config(region_name=NEXITY_COGNITO_REGION)
+            )
+
+        # Request access token
+        client = await loop.run_in_executor(None, _get_client)
+
+        aws = WarrantLite(
+            username=username,
+            password=password,
+            pool_id=NEXITY_COGNITO_USER_POOL,
+            client_id=NEXITY_COGNITO_CLIENT_ID,
+            client=client,
+        )
+
+        try:
+            tokens = await loop.run_in_executor(None, aws.authenticate_user)
+        except Exception as error:
+            raise NexityBadCredentialsException() from error
+
+        async with self.session.get(
+            NEXITY_API + "/deploy/api/v1/domotic/token",
+            headers={
+                "Authorization": tokens["AuthenticationResult"]["IdToken"],
+            },
+        ) as response:
+            token = await response.json()
+
+            if "token" not in token:
+                raise NexityServiceException("No Nexity SSO token provided.")
+
+            sso_token = cast(str, token["token"])
+
+        user_id = username.replace("@", "_-_")  # Replace @ for _-_
+        payload = {"ssoToken": sso_token, "userId": user_id}
+
+        post_response = await self.post("login", data=payload)
+
+        return "success" in post_response