diff --git a/README.md b/README.md index bc9bace1..197db771 100644 --- a/README.md +++ b/README.md @@ -20,11 +20,11 @@ Full documentation is available at **[imicknl.github.io/python-overkiz-api](http | --- | :-: | :-: | | Atlantic Cozytouch | ✓ | | | Bouygues Flexom | ✓ | | -| Brandt Smart Control † | ✓ | | +| Brandt Smart Control | ✓ | | | Hexaom HexaConnect | ✓ | | | Hitachi Hi Kumo | ✓ | | | Nexity Eugénie | ✓ | | -| Rexel Energeasy Connect ‡ | ✓ | ✓ | +| Rexel Energeasy Connect † | ✓ | ✓ | | Sauter Cozytouch | ✓ | | | Simu LiveIn2 | ✓ | | | Somfy | ✓ | ✓ | @@ -33,9 +33,7 @@ Full documentation is available at **[imicknl.github.io/python-overkiz-api](http Local API availability depends on your specific gateway. See the [Getting started guide](https://imicknl.github.io/python-overkiz-api/getting-started/) for the supported gateways and setup. -† _This server's authentication method isn't supported yet. To use it, obtain an access token (by sniffing the original app) and create a local user on the Overkiz API platform._ - -‡ _The cloud API requires OAuth credentials provided by Rexel; the local API uses a token from the EConnect app instead._ +† _The cloud API requires OAuth credentials provided by Rexel; the local API uses a token from the EConnect app instead._ ### Somfy diff --git a/docs/index.md b/docs/index.md index bc4c25f2..3d4c7ad9 100644 --- a/docs/index.md +++ b/docs/index.md @@ -26,11 +26,11 @@ pyOverkiz is an async Python library for interacting with Overkiz-based platform | --- | :-: | :-: | | Atlantic Cozytouch | ✓ | | | Bouygues Flexom | ✓ | | -| Brandt Smart Control † | ✓ | | +| Brandt Smart Control | ✓ | | | Hexaom HexaConnect | ✓ | | | Hitachi Hi Kumo | ✓ | | | Nexity Eugénie | ✓ | | -| Rexel Energeasy Connect ‡ | ✓ | ✓ | +| Rexel Energeasy Connect † | ✓ | ✓ | | Sauter Cozytouch | ✓ | | | Simu LiveIn2 | ✓ | | | Somfy | ✓ | ✓ | @@ -39,9 +39,7 @@ pyOverkiz is an async Python library for interacting with Overkiz-based platform Local API availability depends on your specific gateway. See the [Getting started guide](getting-started.md) for the supported gateways and setup. -† _This server's authentication method isn't supported yet. To use it, obtain an access token (by sniffing the original app) and create a local user on the Overkiz API platform._ - -‡ _The cloud API requires OAuth credentials provided by Rexel; the local API uses a token from the EConnect app instead._ +† _The cloud API requires OAuth credentials provided by Rexel; the local API uses a token from the EConnect app instead._ ### Somfy diff --git a/pyoverkiz/auth/factory.py b/pyoverkiz/auth/factory.py index 8d39716c..81912a07 100644 --- a/pyoverkiz/auth/factory.py +++ b/pyoverkiz/auth/factory.py @@ -17,6 +17,7 @@ from pyoverkiz.auth.strategies import ( AuthStrategy, BearerTokenAuthStrategy, + BrandtAuthStrategy, CozytouchAuthStrategy, LocalTokenAuthStrategy, NexityAuthStrategy, @@ -81,6 +82,14 @@ def build_auth_strategy( ssl_context, ) + if server == Server.BRANDT: + return BrandtAuthStrategy( + _ensure_credentials(credentials, UsernamePasswordCredentials), + session, + server_config, + ssl_context, + ) + if server == Server.REXEL: if isinstance(credentials, RexelTokenCredentials): return RexelTokenAuthStrategy( diff --git a/pyoverkiz/auth/strategies.py b/pyoverkiz/auth/strategies.py index ce9530a2..972c20fc 100644 --- a/pyoverkiz/auth/strategies.py +++ b/pyoverkiz/auth/strategies.py @@ -25,6 +25,8 @@ UsernamePasswordCredentials, ) from pyoverkiz.const import ( + BRANDT_MIDDLEWARE_API, + BRANDT_PARTNER, COZYTOUCH_ATLANTIC_API, COZYTOUCH_CLIENT_ID, NEXITY_API, @@ -43,6 +45,8 @@ ) from pyoverkiz.exceptions import ( BadCredentialsError, + BrandtBadCredentialsError, + BrandtServiceError, CozyTouchBadCredentialsError, CozyTouchServiceError, InvalidTokenError, @@ -264,6 +268,59 @@ async def login(self) -> None: await self._post_login({"jwt": jwt}) +class BrandtAuthStrategy(SessionLoginStrategy): + """Authentication strategy for Brandt Smart Control. + + Brandt fronts Overkiz with a cookie-session middleware: authenticate + against smartcontrol-app.com, fetch a JWT using the resulting session + cookie, then log in to the Overkiz cloud with that JWT. The shared + aiohttp session carries the cookie between the two middleware calls. + """ + + async def login(self) -> None: + """Perform the Brandt middleware login, then Overkiz JWT login.""" + # 1) Middleware session login (sets the devise session cookie). + async with self.session.post( + f"{BRANDT_MIDDLEWARE_API}/api/v1/sessions.json", + json={ + "client": { + "email": self.credentials.username, + "password": self.credentials.password, + "partner": BRANDT_PARTNER, + } + }, + ssl=self._ssl, + ) as response: + if response.status >= HTTPStatus.BAD_REQUEST: + message = "Login failed: bad credentials" + try: + body = await response.json() + errors = body.get("error") + if errors: + message = errors[0] + except ValueError: + pass + raise BrandtBadCredentialsError(message) + + # 2) Fetch the JWT, authenticated purely by the session cookie. + async with self.session.get( + f"{BRANDT_MIDDLEWARE_API}/api/v1/profile/jwt.json", + ssl=self._ssl, + ) as response: + if response.status >= HTTPStatus.BAD_REQUEST: + raise BrandtServiceError( + f"Brandt JWT request failed: {response.status}" + ) + body = await response.json() + jwt = body.get("client", {}).get("jwt") + + if not jwt: + raise BrandtServiceError("No Brandt JWT token provided.") + + # 3) Overkiz cloud login with the JWT only (no apiKey/applicationId). + await self._post_login({"jwt": jwt}) + + class NexityAuthStrategy(SessionLoginStrategy): """Authentication strategy using Nexity session-based login.""" diff --git a/pyoverkiz/const.py b/pyoverkiz/const.py index e2e7e4bb..3fd8ce27 100644 --- a/pyoverkiz/const.py +++ b/pyoverkiz/const.py @@ -42,6 +42,10 @@ # OAuth client secrets are public by design (embedded in mobile apps) SOMFY_CLIENT_SECRET = "12k73w1n540g8o4cokg0cw84cog840k84cwggscwg884004kgk" # noqa: S105 +# Brandt Smart Control middleware (cookie-session Rails API in front of Overkiz) +BRANDT_MIDDLEWARE_API = "https://www.smartcontrol-app.com" +BRANDT_PARTNER = "brandt-electromenager" + LOCAL_API_PATH = "/enduser-mobile-web/1/enduserAPI/" SERVERS_WITH_LOCAL_API = [ diff --git a/pyoverkiz/exceptions.py b/pyoverkiz/exceptions.py index 05c7756d..a416f007 100644 --- a/pyoverkiz/exceptions.py +++ b/pyoverkiz/exceptions.py @@ -153,3 +153,11 @@ class SomfyBadCredentialsError(BadCredentialsError): class SomfyServiceError(BaseOverkizError): """Raised when an error occurs while communicating with the Somfy API.""" + + +class BrandtBadCredentialsError(BadCredentialsError): + """Raised when the Brandt middleware rejects the username/password.""" + + +class BrandtServiceError(BaseOverkizError): + """Raised when the Brandt middleware returns an unexpected response.""" diff --git a/tests/test_auth.py b/tests/test_auth.py index 09dc2428..e89384a3 100644 --- a/tests/test_auth.py +++ b/tests/test_auth.py @@ -139,6 +139,22 @@ def test_rexel_token_credentials_requires_a_token_source(self): RexelTokenCredentials() +def test_brandt_constants_and_exceptions(): + """Brandt middleware constants and exceptions are exported correctly.""" + from pyoverkiz.const import BRANDT_MIDDLEWARE_API, BRANDT_PARTNER + from pyoverkiz.exceptions import ( + BadCredentialsError, + BaseOverkizError, + BrandtBadCredentialsError, + BrandtServiceError, + ) + + assert BRANDT_MIDDLEWARE_API == "https://www.smartcontrol-app.com" + assert BRANDT_PARTNER == "brandt-electromenager" + assert issubclass(BrandtBadCredentialsError, BadCredentialsError) + assert issubclass(BrandtServiceError, BaseOverkizError) + + class TestAuthFactory: """Test authentication factory functions.""" @@ -228,6 +244,30 @@ async def test_build_auth_strategy_cozytouch(self): assert isinstance(strategy, CozytouchAuthStrategy) + @pytest.mark.asyncio + async def test_build_auth_strategy_brandt(self): + """Test building Brandt auth strategy.""" + from pyoverkiz.auth.strategies import BrandtAuthStrategy + + server_config = ServerConfig( + server=Server.BRANDT, + name="Brandt Smart Control", + endpoint="https://ha3-1.overkiz.com/enduser-mobile-web/enduserAPI/", + manufacturer="Brandt", + api_type=APIType.CLOUD, + ) + credentials = UsernamePasswordCredentials("user", "pass") + session = AsyncMock(spec=ClientSession) + + strategy = build_auth_strategy( + server_config=server_config, + credentials=credentials, + session=session, + ssl_context=True, + ) + + assert isinstance(strategy, BrandtAuthStrategy) + @pytest.mark.asyncio async def test_build_auth_strategy_nexity(self): """Test building Nexity auth strategy.""" @@ -546,6 +586,112 @@ async def test_auth_headers_no_token(self): assert headers == {} +class TestBrandtAuthStrategy: + """Test BrandtAuthStrategy.""" + + @staticmethod + def _server_config(): + return ServerConfig( + server=Server.BRANDT, + name="Brandt Smart Control", + endpoint="https://ha3-1.overkiz.com/enduser-mobile-web/enduserAPI/", + manufacturer="Brandt", + api_type=APIType.CLOUD, + ) + + @staticmethod + def _json_response(status, payload): + resp = MagicMock() + resp.status = status + resp.json = AsyncMock(return_value=payload) + resp.__aenter__ = AsyncMock(return_value=resp) + resp.__aexit__ = AsyncMock(return_value=None) + return resp + + @pytest.mark.asyncio + async def test_login_success_flow(self): + """Full flow: sessions.json -> profile/jwt.json -> Overkiz login.""" + from pyoverkiz.auth.strategies import BrandtAuthStrategy + + session = AsyncMock(spec=ClientSession) + # 1) POST sessions.json 2) Overkiz POST login (inherited _post_login) + session.post = MagicMock( + side_effect=[ + self._json_response(200, {"client": {"email": "a@b.c"}}), + self._json_response(200, {"success": True}), + ] + ) + # GET profile/jwt.json + session.get = MagicMock( + return_value=self._json_response(200, {"client": {"jwt": "the.jwt.token"}}) + ) + + credentials = UsernamePasswordCredentials("a@b.c", "pass") + strategy = BrandtAuthStrategy(credentials, session, self._server_config(), True) + await strategy.login() + + # First POST is to the middleware with the partner field. + first_post = session.post.call_args_list[0] + assert ( + first_post.args[0] + == "https://www.smartcontrol-app.com/api/v1/sessions.json" + ) + assert first_post.kwargs["json"] == { + "client": { + "email": "a@b.c", + "password": "pass", + "partner": "brandt-electromenager", + } + } + # JWT fetched from the cookie-authenticated GET. + session.get.assert_called_once_with( + "https://www.smartcontrol-app.com/api/v1/profile/jwt.json", ssl=True + ) + # Final POST is the Overkiz login carrying only the jwt. + last_post = session.post.call_args_list[-1] + assert last_post.args[0] == ( + "https://ha3-1.overkiz.com/enduser-mobile-web/enduserAPI/login" + ) + assert last_post.kwargs["data"] == {"jwt": "the.jwt.token"} + + @pytest.mark.asyncio + async def test_login_bad_credentials(self): + """A wrong-password 400 from sessions.json raises BrandtBadCredentialsError. + + Pins the real middleware contract: status 400 with an ``error`` array; + the first element is surfaced as the exception message. + """ + from pyoverkiz.auth.strategies import BrandtAuthStrategy + from pyoverkiz.exceptions import BrandtBadCredentialsError + + session = AsyncMock(spec=ClientSession) + session.post = MagicMock( + return_value=self._json_response( + 400, {"error": ["Password wrong password"], "status": 400} + ) + ) + + credentials = UsernamePasswordCredentials("a@b.c", "wrong") + strategy = BrandtAuthStrategy(credentials, session, self._server_config(), True) + with pytest.raises(BrandtBadCredentialsError, match="Password wrong password"): + await strategy.login() + + @pytest.mark.asyncio + async def test_login_missing_jwt_raises_service_error(self): + """Login OK but no jwt in the profile response -> BrandtServiceError.""" + from pyoverkiz.auth.strategies import BrandtAuthStrategy + from pyoverkiz.exceptions import BrandtServiceError + + session = AsyncMock(spec=ClientSession) + session.post = MagicMock(return_value=self._json_response(200, {"client": {}})) + session.get = MagicMock(return_value=self._json_response(200, {"client": {}})) + + credentials = UsernamePasswordCredentials("a@b.c", "pass") + strategy = BrandtAuthStrategy(credentials, session, self._server_config(), True) + with pytest.raises(BrandtServiceError): + await strategy.login() + + class TestBearerTokenAuthStrategy: """Test BearerTokenAuthStrategy."""