maint: Add protections in line with base

custom_components/ms365_mail/__init__.py

@@ -4,17 +4,16 @@
 
 from homeassistant.core import HomeAssistant
 from homeassistant.helpers import issue_registry as ir
-from O365 import Account, FileSystemTokenBackend
 from oauthlib.oauth2.rfc6749.errors import InvalidClientError
 
 from .const import (
     CONF_CLIENT_ID,
     CONF_CLIENT_SECRET,
     CONF_ENTITY_NAME,
     CONF_SHARED_MAILBOX,
-    CONST_UTC_TIMEZONE,
 )
 from .helpers.config_entry import MS365ConfigEntry, MS365Data
+from .integration import setup_integration
 from .integration.const_integration import DOMAIN, PLATFORMS
 from .integration.permissions_integration import Permissions
 from .integration.setup_integration import async_do_setup
@@ -36,8 +35,8 @@ async def async_setup_entry(hass: HomeAssistant, entry: MS365ConfigEntry):
     perms = Permissions(hass, entry.data)
     permissions, failed_permissions = await perms.async_check_authorizations()  # pylint: disable=unused-variable
     if permissions is True:
-        account, is_authenticated = await hass.async_add_executor_job(
-            _try_authentication, perms, credentials, main_resource
+        account, is_authenticated, auth_error = await hass.async_add_executor_job(  # pylint: disable=unused-variable
+            perms.try_authentication, credentials, main_resource
         )
     else:
         is_authenticated = False
@@ -81,22 +80,13 @@ async def async_reload_entry(hass: HomeAssistant, entry: MS365ConfigEntry) -> No
         await hass.config_entries.async_reload(entry.entry_id)
 
 
-def _try_authentication(perms, credentials, main_resource):
-    _LOGGER.debug("Setup token")
-    token_backend = FileSystemTokenBackend(
-        token_path=perms.token_path,
-        token_filename=perms.token_filename,
-    )
-
-    _LOGGER.debug("Setup account")
-    account = Account(
-        credentials,
-        token_backend=token_backend,
-        timezone=CONST_UTC_TIMEZONE,
-        main_resource=main_resource,
-    )
-
-    return account, account.is_authenticated
+async def async_remove_entry(hass: HomeAssistant, entry: MS365ConfigEntry) -> None:
+    """Handle removal of an entry."""
+    perms = Permissions(hass, entry.data)
+    await hass.async_add_executor_job(perms.delete_token)
+    if not hasattr(setup_integration, "async_integration_remove_entry"):
+        return
+    await setup_integration.async_integration_remove_entry(hass, entry)
 
 
 async def _async_check_token(hass, account, entity_name):

custom_components/ms365_mail/classes/permissions.py

@@ -5,8 +5,11 @@
 import os
 from copy import deepcopy
 
+from O365 import Account, FileSystemTokenBackend
+
 from ..const import (
     CONF_ENTITY_NAME,
+    CONST_UTC_TIMEZONE,
     MS365_STORAGE_TOKEN,
     PERM_OFFLINE_ACCESS,
     TOKEN_FILE_CORRUPTED,
@@ -29,9 +32,14 @@ def __init__(self, hass, config):
         self._config = config
 
         self._requested_permissions = []
+        self._permissions = []
         self.token_filename = self.build_token_filename()
         self.token_path = build_config_file_path(self._hass, MS365_STORAGE_TOKEN)
-        self._permissions = []
+        _LOGGER.debug("Setup token")
+        self.token_backend = FileSystemTokenBackend(
+            token_path=self.token_path,
+            token_filename=self.token_filename,
+        )
 
     @property
     def requested_permissions(self):
@@ -42,6 +50,22 @@ def permissions(self):
         """Return the permission set."""
         return self._permissions
 
+    def try_authentication(self, credentials, main_resource):
+        """Try authenticating to O365."""
+        _LOGGER.debug("Setup account")
+        account = Account(
+            credentials,
+            token_backend=self.token_backend,
+            timezone=CONST_UTC_TIMEZONE,
+            main_resource=main_resource,
+        )
+        try:
+            return account, account.is_authenticated, False
+
+        except json.decoder.JSONDecodeError as err:
+            _LOGGER.error("Error authenticating - JSONDecodeError - %s", err)
+            return account, False, err
+
     async def async_check_authorizations(self):
         """Report on permissions status."""
         self._permissions = await self._hass.async_add_executor_job(
@@ -141,3 +165,9 @@ def _get_permissions(self):
             return TOKEN_FILE_CORRUPTED
 
         return permissions
+
+    def delete_token(self):
+        """Delete the token."""
+        full_token_path = os.path.join(self.token_path, self.token_filename)
+        if os.path.exists(full_token_path):
+            os.remove(full_token_path)

custom_components/ms365_mail/config_flow.py

@@ -1,7 +1,6 @@
 """Configuration flow for the skyq platform."""
 
 import functools as ft
-import json
 import logging
 from collections.abc import Mapping
 from typing import Any, Self
@@ -19,7 +18,6 @@
 from homeassistant.data_entry_flow import FlowResult
 from homeassistant.helpers import issue_registry as ir
 from homeassistant.helpers.network import get_url
-from O365 import Account, FileSystemTokenBackend
 
 from .const import (
     AUTH_CALLBACK_NAME,
@@ -33,7 +31,7 @@
     CONF_FAILED_PERMISSIONS,
     CONF_SHARED_MAILBOX,
     CONF_URL,
-    CONST_UTC_TIMEZONE,
+    ERROR_INVALID_SHARED_MAILBOX,
     TOKEN_FILE_CORRUPTED,
     TOKEN_FILE_MISSING,
     TOKEN_FILE_PERMISSIONS,
@@ -48,10 +46,7 @@
 from .integration.const_integration import DOMAIN
 from .integration.permissions_integration import Permissions
 from .integration.schema_integration import CONFIG_SCHEMA_INTEGRATION
-from .schema import (
-    CONFIG_SCHEMA,
-    REQUEST_AUTHORIZATION_DEFAULT_SCHEMA,
-)
+from .schema import CONFIG_SCHEMA, REQUEST_AUTHORIZATION_DEFAULT_SCHEMA
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -117,8 +112,7 @@ async def async_step_user(self, user_input=None):
                 is_authenticated,
                 auth_error,
             ) = await self.hass.async_add_executor_job(
-                self._try_authentication,
-                self._permissions,
+                self._permissions.try_authentication,
                 credentials,
                 main_resource,
             )
@@ -242,35 +236,39 @@ async def _async_validate_response(self, user_input):
             errors[CONF_URL] = "token_file_error"
             return errors
 
+        credentials = (
+            self._user_input.get(CONF_CLIENT_ID),
+            self._user_input.get(CONF_CLIENT_SECRET),
+        )
+
+        main_resource = self._user_input.get(CONF_SHARED_MAILBOX)
+        (
+            self._account,
+            is_authenticated,  # pylint: disable=unused-variable
+            auth_error,  # pylint: disable=unused-variable
+        ) = await self.hass.async_add_executor_job(
+            self._permissions.try_authentication,
+            credentials,
+            main_resource,
+        )
+        if (
+            hasattr(self._account, "current_username")
+            and self._account.current_username
+            and self._account.current_username == self._account.main_resource
+        ):
+            self._user_input[CONF_SHARED_MAILBOX] = None
+            _LOGGER.info(ERROR_INVALID_SHARED_MAILBOX, self._account.current_username)
+
         (
             permissions,
             self._failed_permissions,
         ) = await self._permissions.async_check_authorizations()
+
         if permissions is not True:
             errors[CONF_URL] = permissions
 
         return errors
 
-    def _try_authentication(self, perms, credentials, main_resource):
-        _LOGGER.debug("Setup token")
-        token_backend = FileSystemTokenBackend(
-            token_path=perms.token_path,
-            token_filename=perms.token_filename,
-        )
-        _LOGGER.debug("Setup account")
-        account = Account(
-            credentials,
-            token_backend=token_backend,
-            timezone=CONST_UTC_TIMEZONE,
-            main_resource=main_resource,
-        )
-        try:
-            return account, account.is_authenticated, False
-
-        except json.decoder.JSONDecodeError as err:
-            _LOGGER.error("Error authenticating - JSONDecodeError - %s", err)
-            return account, False, err
-
     async def async_step_reconfigure(
         self,
         user_input: Mapping[str, Any] | None = None,  # pylint: disable=unused-argument

custom_components/ms365_mail/const.py

@@ -35,6 +35,10 @@
 PERM_USER_READ = "User.Read"
 PERM_SHARED = ".Shared"
 
+ERROR_INVALID_SHARED_MAILBOX = (
+    "Login email address '%s' should not be "
+    + "entered as shared email address, config attribute removed."
+)
 
 TOKEN_FILENAME = "{0}{1}.token"  # nosec
 TOKEN_FILE_CORRUPTED = "corrupted"

custom_components/ms365_mail/helpers/utils.py

@@ -1,5 +1,7 @@
 """Utilities processes."""
 
+from copy import deepcopy
+
 from bs4 import BeautifulSoup
 from homeassistant.helpers.entity import async_generate_entity_id
 
@@ -37,3 +39,8 @@ def build_entity_id(hass, entity_id_format, name):
         name,
         hass=hass,
     )
+
+
+def shared_permission_build(permission, shared):
+    """Build the shared permission."""
+    return f"{deepcopy(permission)}.Shared" if shared else permission

tests/integration/const_integration.py

@@ -25,6 +25,7 @@
 }
 BASE_TOKEN_PERMS = "Mail.Read"
 BASE_MISSING_PERMS = BASE_TOKEN_PERMS
+SHARED_TOKEN_PERMS = "Mail.Read.Shared"
 UPDATE_TOKEN_PERMS = "Mail.Read Mail.Send MailboxSettings.ReadWrite"
 UPDATE_OPTIONS = {"enable_update": True}
 

tests/integration/test_config_flow.py

@@ -1,8 +1,14 @@
 # pylint: disable=line-too-long, unused-argument
 """Test the config flow."""
 
+from copy import deepcopy
+from unittest.mock import MagicMock, patch
+
+import pytest
+from homeassistant import config_entries
 from homeassistant.core import HomeAssistant
 from homeassistant.data_entry_flow import FlowResultType
+from requests_mock import Mocker
 
 from custom_components.ms365_mail.integration.const_integration import (
     CONF_HAS_ATTACHMENT,
@@ -14,7 +20,14 @@
 )
 
 from ..helpers.mock_config_entry import MS365MockConfigEntry
-from ..helpers.utils import get_schema_default
+from ..helpers.utils import build_token_url, get_schema_default, mock_token
+from .const_integration import (
+    AUTH_CALLBACK_PATH_DEFAULT,
+    BASE_CONFIG_ENTRY,
+    DOMAIN,
+    SHARED_TOKEN_PERMS,
+)
+from .helpers_integration.mocks import MS365MOCKS
 
 
 async def test_options_flow(
@@ -76,3 +89,48 @@ async def test_options_flow_empty(
     assert CONF_IMPORTANCE not in result["data"]
     assert CONF_HAS_ATTACHMENT not in result["data"]
     assert CONF_IS_UNREAD not in result["data"]
+
+
+async def test_shared_email_invalid(
+    hass: HomeAssistant,
+    requests_mock: Mocker,
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """Test for invalid shared mailbox."""
+    mock_token(requests_mock, SHARED_TOKEN_PERMS)
+    MS365MOCKS.standard_mocks(requests_mock)
+
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN, context={"source": config_entries.SOURCE_USER}
+    )
+
+    user_input = deepcopy(BASE_CONFIG_ENTRY)
+    email = "[email protected]"
+    user_input["shared_mailbox"] = email
+    result = await hass.config_entries.flow.async_configure(
+        result["flow_id"],
+        user_input=user_input,
+    )
+
+    with patch(
+        f"custom_components.{DOMAIN}.classes.permissions.Account",
+        return_value=mock_account(email),
+    ):
+        result = await hass.config_entries.flow.async_configure(
+            result["flow_id"],
+            user_input={
+                "url": build_token_url(result, AUTH_CALLBACK_PATH_DEFAULT),
+            },
+        )
+
+    assert result["type"] is FlowResultType.CREATE_ENTRY
+
+    assert (
+        f"Login email address '{email}' should not be entered as shared email address, config attribute removed"
+        in caplog.text
+    )
+
+
+def mock_account(email):
+    """Mock the account."""
+    return MagicMock(is_authenticated=True, current_username=email, main_resource=email)

tests/test_init.py

@@ -9,8 +9,9 @@
 from oauthlib.oauth2.rfc6749.errors import InvalidClientError
 from requests_mock import Mocker
 
+from .const import ENTITY_NAME, TOKEN_LOCATION
 from .helpers.mock_config_entry import MS365MockConfigEntry
-from .integration.const_integration import FULL_INIT_ENTITY_NO
+from .integration.const_integration import DOMAIN, FULL_INIT_ENTITY_NO
 from .integration.helpers_integration.mocks import MS365MOCKS
 
 
@@ -72,3 +73,17 @@ async def test_invalid_client_2(
         await hass.config_entries.async_setup(base_config_entry.entry_id)
     await hass.async_block_till_done()
     assert "Token error for account" in caplog.text
+
+
+async def test_remove_entry(
+    tmp_path,
+    setup_base_integration,
+    hass: HomeAssistant,
+    base_config_entry: MS365MockConfigEntry,
+):
+    """Test removal of entry."""
+
+    assert await hass.config_entries.async_remove(base_config_entry.entry_id)
+    await hass.async_block_till_done()
+    filename = tmp_path / TOKEN_LOCATION / f"{DOMAIN}_{ENTITY_NAME}.token"
+    assert not filename.is_file()