Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 3 additions & 5 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -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 | ✓ | ✓ |
Expand All @@ -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

Expand Down
8 changes: 3 additions & 5 deletions docs/index.md
Original file line number Diff line number Diff line change
Expand Up @@ -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 | ✓ | ✓ |
Expand All @@ -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

Expand Down
9 changes: 9 additions & 0 deletions pyoverkiz/auth/factory.py
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,7 @@
from pyoverkiz.auth.strategies import (
AuthStrategy,
BearerTokenAuthStrategy,
BrandtAuthStrategy,
CozytouchAuthStrategy,
LocalTokenAuthStrategy,
NexityAuthStrategy,
Expand Down Expand Up @@ -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(
Expand Down
57 changes: 57 additions & 0 deletions pyoverkiz/auth/strategies.py
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,8 @@
UsernamePasswordCredentials,
)
from pyoverkiz.const import (
BRANDT_MIDDLEWARE_API,
BRANDT_PARTNER,
COZYTOUCH_ATLANTIC_API,
COZYTOUCH_CLIENT_ID,
NEXITY_API,
Expand All @@ -43,6 +45,8 @@
)
from pyoverkiz.exceptions import (
BadCredentialsError,
BrandtBadCredentialsError,
BrandtServiceError,
CozyTouchBadCredentialsError,
CozyTouchServiceError,
InvalidTokenError,
Expand Down Expand Up @@ -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."""

Expand Down
4 changes: 4 additions & 0 deletions pyoverkiz/const.py
Original file line number Diff line number Diff line change
Expand Up @@ -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 = [
Expand Down
8 changes: 8 additions & 0 deletions pyoverkiz/exceptions.py
Original file line number Diff line number Diff line change
Expand Up @@ -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."""
146 changes: 146 additions & 0 deletions tests/test_auth.py
Original file line number Diff line number Diff line change
Expand Up @@ -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."""

Expand Down Expand Up @@ -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."""
Expand Down Expand Up @@ -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."""

Expand Down