"""evohome_cli - a credentials manager for the evohome CLI.""" from __future__ import annotations import json import logging import os from datetime import UTC, datetime as dt, timedelta as td from pathlib import Path from typing import TYPE_CHECKING, Any, Final, NotRequired, TypedDict import aiofiles import keyring import keyring.errors from evohomeasync.auth import ( SZ_SESSION_ID, SZ_SESSION_ID_EXPIRES, AbstractSessionManager, ) from evohomeasync2.auth import ( SZ_ACCESS_TOKEN, SZ_ACCESS_TOKEN_EXPIRES, AbstractTokenManager, ) if TYPE_CHECKING: from evohomeasync.auth import SessionIdEntryT from evohomeasync2.auth import AccessTokenEntryT _CACHE_PATH: Final = Path.home() / ".config" / "evohome" / ".evo-cache.tmp~" KEYRING_SERVICE_KEY: Final = "evohome-client" KEYRING_USERNAME_KEY: Final = "username" _LOGGER: Final = logging.getLogger(__name__) # Keyring functions are synchronous: but this is CLI, not library code, so is OK. def is_keyring_available() -> bool: """Return True if a working keyring backend is available.""" import keyring.backends.fail # noqa: PLC0415 return not isinstance(keyring.get_keyring(), keyring.backends.fail.Keyring) def get_password_from_keyring(username: str) -> str | None: """Retrieve the TCC password for the given username from the system keyring.""" try: return keyring.get_password(KEYRING_SERVICE_KEY, username) except keyring.errors.KeyringError as err: _LOGGER.debug("Failed to retrieve password from keyring: %s", err) return None def save_password_to_keyring(username: str, password: str) -> None: """Save the TCC password for the given username to the system keyring.""" try: keyring.set_password(KEYRING_SERVICE_KEY, username, password) except keyring.errors.KeyringError as err: _LOGGER.debug("Failed to save password to keyring: %s", err) def get_username_from_keyring() -> str | None: """Retrieve the stored default TCC username from the system keyring.""" try: return keyring.get_password(KEYRING_SERVICE_KEY, KEYRING_USERNAME_KEY) except keyring.errors.KeyringError as err: _LOGGER.debug("Failed to retrieve username from keyring: %s", err) return None def save_username_to_keyring(username: str) -> None: """Save the TCC username as the default in the system keyring.""" try: keyring.set_password(KEYRING_SERVICE_KEY, KEYRING_USERNAME_KEY, username) except keyring.errors.KeyringError as err: _LOGGER.debug("Failed to save username to keyring: %s", err) def delete_password_from_keyring(username: str) -> bool: """Remove the TCC password for the given username from the system keyring.""" try: keyring.delete_password(KEYRING_SERVICE_KEY, username) except keyring.errors.KeyringError as err: _LOGGER.debug("Failed to delete password from keyring: %s", err) return False return True def delete_username_from_keyring() -> bool: """Remove the stored default TCC username from the system keyring.""" try: keyring.delete_password(KEYRING_SERVICE_KEY, KEYRING_USERNAME_KEY) except keyring.errors.KeyringError as err: _LOGGER.debug("Failed to delete username from keyring: %s", err) return False return True class UserEntryT(TypedDict): access_token: NotRequired[AccessTokenEntryT] session_id: NotRequired[SessionIdEntryT] CacheDataT = dict[str, UserEntryT] # str is the client_id """ { "username@gmail.com": { "access_token": { "access_token": "iiuyz2-...", "access_token_expires": "2024-09-24T10:25:12+01:00", "refresh_token": "dfsadgf..." }, "session_id": { "session_id": "94A76CB4-8BD4-4600-AAE4-...", "session_id_expires": "2024-09-24T10:25:12+01:00" } }, "username@email.com": {} } """ class TokenCacheManager(AbstractTokenManager, AbstractSessionManager): """A credentials manager that uses a file to cache the tokens.""" def __init__( self, *args: Any, cache_path: Path = _CACHE_PATH, **kwargs: Any ) -> None: """Initialise the credentials manager (for access_token & session_id).""" # ensure default logger as we've merged the two ABCs kwargs["logger"] = kwargs.get("logger") or logging.getLogger(__name__) super().__init__(*args, **kwargs) self._cache_path: Final = cache_path @property def cache_path(self) -> Path: """Return the token cache path.""" return self._cache_path @staticmethod def _clean_cache(old_cache: CacheDataT) -> CacheDataT: """Return a copy of a cache with any expired data removed.""" new_cache: CacheDataT = {} dt_now = (dt.now(tz=UTC) + td(seconds=15)).isoformat() for user_id, entry in old_cache.items(): user_data: UserEntryT = {} if (t := entry.get(SZ_ACCESS_TOKEN)) and t[ SZ_ACCESS_TOKEN_EXPIRES ] > dt_now: user_data[SZ_ACCESS_TOKEN] = t # session_id is not used by evohomeasync2 if (s := entry.get(SZ_SESSION_ID)) and s[SZ_SESSION_ID_EXPIRES] > dt_now: user_data[SZ_SESSION_ID] = s if user_data: new_cache[user_id] = user_data return new_cache # could be Falsey async def _read_cache_from_file(self) -> CacheDataT: """Return a copy of the cache as read from file.""" try: async with aiofiles.open(self.cache_path) as fp: content = await fp.read() or "{}" except FileNotFoundError: return {} cache: CacheDataT = json.loads(content) return cache async def _write_cache_to_file(self, cache: CacheDataT) -> None: """Write the supplied cache to file.""" content = json.dumps(cache, indent=4) if not self.cache_path.parent.exists(): self.cache_path.parent.mkdir(mode=0o700, parents=True) fd = os.open(self.cache_path, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600) async with aiofiles.open(fd, "w", closefd=True) as fp: await fp.write(content) async def load_from_cache(self) -> None: """Load the user entry from the cache.""" cache: CacheDataT = await self._read_cache_from_file() await self._load_access_token(cache=cache) await self._load_session_id(cache=cache) async def _load_access_token(self, cache: CacheDataT | None = None) -> None: """Load the (serialized) auth tokens from the cache.""" cache = cache or await self._read_cache_from_file() entry: UserEntryT | None = cache.get(self.client_id) if not entry: return tokens: AccessTokenEntryT | None = entry.get(SZ_ACCESS_TOKEN) if not tokens: return # if not self._access_token: # not needed as dt.min is sentinel for this if self._access_token_expires.isoformat() < tokens[SZ_ACCESS_TOKEN_EXPIRES]: self._import_access_token(tokens) async def _load_session_id(self, cache: CacheDataT | None = None) -> None: """Load the (serialized) session id from the cache.""" cache = cache or await self._read_cache_from_file() entry: UserEntryT | None = cache.get(self.client_id) if not entry: return session: SessionIdEntryT | None = entry.get(SZ_SESSION_ID) if not session: return # if not self._session_id: # not needed as dt.min is sentinel for this if self._session_id_expires.isoformat() < session[SZ_SESSION_ID_EXPIRES]: self._import_session_id(session) async def save_to_cache(self) -> None: """Save the user entry to the cache.""" await self.save_access_token() await self.save_session_id() async def save_access_token(self) -> None: """Save the (serialized) access token to the cache. Includes the access token expiry datetime, and the refresh token. """ cache: CacheDataT = await self._read_cache_from_file() if self.client_id not in cache: cache[self.client_id] = {} cache[self.client_id][SZ_ACCESS_TOKEN] = self._export_access_token() await self._write_cache_to_file(self._clean_cache(cache)) async def save_session_id(self) -> None: """Save the (serialized) session id to the cache. Includes the session id expiry datetime. """ cache: CacheDataT = await self._read_cache_from_file() if self.client_id not in cache: cache[self.client_id] = {} cache[self.client_id][SZ_SESSION_ID] = self._export_session_id() await self._write_cache_to_file(self._clean_cache(cache))