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
12 changes: 12 additions & 0 deletions fritz_advanced_thermostat/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -390,6 +390,18 @@ def get_thermostats(self) -> set:
self._thermostats.add(dev_name)
return self._thermostats

def refresh_sid(self) -> None:
"""Re-authenticate with the Fritz!Box to obtain a fresh session ID (SID).

This is useful for long-running applications where the SID may expire
due to inactivity (~20 minutes).

Raises:
FritzAdvancedThermostatConnectionError: If re-authentication fails.

"""
self._fritz_conn.refresh_sid()

def reload_thermostat_data(self) -> None:
"""Force a reload of all thermostat data from the Fritz!Box."""
self._generate_thermostat_data(force_reload=True)
16 changes: 15 additions & 1 deletion fritz_advanced_thermostat/utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -29,6 +29,8 @@ def __init__(self, prefixed_host: str, max_retries: int, timeout: int, ssl_verif
self._ssl_verify = ssl_verify
self._logger = logging.getLogger("FritzAdvancedThermostatLogger")
self._sid = None
self._user = None
self._password = None

def _generate_headers(self, data: dict) -> dict:
self._logger.debug("Generating headers for the request.")
Expand All @@ -47,7 +49,7 @@ def _generate_headers(self, data: dict) -> dict:
self._logger.debug("Headers generated: %s", headers)
return headers

def post_req(self, payload: dict, site: str) -> str:
def post_req(self, payload: dict, site: str, _sid_refreshed: bool = False) -> str:
"""Send a POST request to the Fritz!Box."""
url = f"{self._prefixed_host}/{site}"
payload = {"sid": self._sid} | payload
Expand Down Expand Up @@ -99,11 +101,19 @@ def post_req(self, payload: dict, site: str) -> str:
self._logger.error("Request failed: %s", err)
raise FritzAdvancedThermostatConnectionError(err)

if '"sid":"0000000000000000"' in response.text and not _sid_refreshed:
self._logger.warning("SID expired, refreshing SID and retrying request.")
self.refresh_sid()
payload.pop("sid")
return self.post_req(payload, site, _sid_refreshed=True)

self._logger.debug("Response received: %s", response.text)
return response.text

def login(self, user: str, password: str) -> None:
"""Authenticate with the Fritz!Box using PBKDF2 challenge-response."""
self._user = user
self._password = password
url = "/".join([self._prefixed_host, f"login_sid.lua?version=2&user={user}"])
response = requests.get(
url, verify=self._ssl_verify, timeout=self._timeout)
Expand Down Expand Up @@ -138,6 +148,10 @@ def login(self, user: str, password: str) -> None:
raise FritzAdvancedThermostatConnectionError(err)
self._sid = sid

def refresh_sid(self) -> None:
"""Re-authenticate with the Fritz!Box to obtain a fresh SID."""
self.login(self._user, self._password)

def get_fritz_os_version(self) -> str:
"""Retrieve the Fritz!OS version from the Fritz!Box."""
payload = {
Expand Down