[WIP] Authorizables and authorization checks

ticgithub/authorizable.py

@@ -0,0 +1,47 @@
+import logging
+_logger = logging.getLogger(__name__)
+
+
+class Authorizable:
+    """Mixin for authorizable objects, to provide authorization checks.
+
+    Subclasses must set the class var ``_AUTHORIZATION_ERROR_CLS`` and must
+    implement :meth:`._check_authorization`.
+
+    Note that ``_AUTHORIZATION_ERROR_CLS`` can be either a single exception
+    class, or a tuple of them. It is only used as part of an ``except``
+    catch.
+    """
+    _AUTHORIZATION_ERROR_CLS = None
+
+    def _check_authorization(self) -> bool:
+        """Check the authorization for this object.
+
+        If authorization succeeds, this should return True. If authorization
+        fails, this should raise ``self._AUTHORIZATION_ERROR_CLS``.
+        """
+        raise NotImplementedError()
+
+    def validate_authorization(self, logger=_logger, label=None):
+        """Validate the authorization for this object.
+
+        This catches the errors raised on authorization failure, ensuring
+        that a failed authorization returns ``False``. This also logs the
+        failures at log level CRITICAL to the given ``logger`` (unless
+        ``logger`` is ``None``).
+
+        Parameters
+        ----------
+        """
+        if label is None:
+            label = self.__class__.__name__
+
+        try:
+            authorized = self._check_authorization()
+        except self._AUTHORIZATION_ERROR_CLS as e:
+            authorized = False
+            if logger is not None:
+                logger.critical(f"{label}: Authorization failed")
+                logger.critical(f"{e.__class__.__name__}: {str(e)}")
+
+        return authorized

ticgithub/bot.py

@@ -5,10 +5,13 @@
 import github
 
 from .issues import Issue, NonTicketIssueError
+from .authorizable import Authorizable
 
 __all__ = ["SMTP", "Bot"]
 
-class SMTP:
+class SMTP(Authorizable):
+    _AUTHORIZATION_ERROR_CLS = smtplib.SMTPException
+
     def __init__(self, user, host, secret, port=465):
         self.user = user
         self.host = host
@@ -20,12 +23,17 @@ def sendmail(self, email, recipients):
             smtp.login(self.user, os.environ.get(self.secret))
             smtp.sendmail(self.user, recipients, email.as_string())
 
+    def _check_authorization(self):
+        with smtplib.SMTP_SSL(self.host, self.port) as smtp:
+            smtp.login(self.user, os.environ.get(self.secret))
+        return True
+
     def __repr__(self):
         return (f"{self.__class__.__name__}('{self.user}:"
                 f"{self.port}")
 
 
-class Bot:
+class Bot(Authorizable):
     def __init__(self, token_secret, repo, smtp=None):
         self.token_secret = token_secret
         self.reponame = repo
@@ -46,6 +54,10 @@ def from_config(cls, config):
                    repo=config['repo'],
                    smtp=smtp)
 
+    def _check_authorization(self):
+        self.repo  # this should raise any errors
+        return True
+
     @property
     def repo(self):
         if self._github is None:

ticgithub/inbox.py

@@ -8,6 +8,8 @@
 from email.header import decode_header
 from email.contentmanager import raw_data_manager
 
+from .authorizable import Authorizable
+
 __all__ = ["Message", "Inbox"]
 
 class Message:
@@ -61,11 +63,13 @@ def get_content(self, content_type_order=None):
         return raw_data_manager.get_content(message)
 
 
-class Inbox:
+class Inbox(Authorizable):
     MESSAGE_CLASS = Message
     FETCH_STR = "RFC822"
     TYPE = "imap"
 
+    _AUTHORIZATION_ERROR_CLS = imaplib.IMAP4.error
+
     def __init__(
         self,
         host,
@@ -90,6 +94,12 @@ def from_config(cls, config):
         kwargs = {k: v for k, v in config.items() if k != "type"}
         return cls(**kwargs)
 
+    def _check_authorization(self):
+        with self.connection:
+            # any errors should be raised by that
+            pass
+        return True
+
     @contextlib.contextmanager
     def connection(self):
         """Get a connection to the IMAP server (use as context manager)."""

ticgithub/tasks/authorization_check.py

@@ -0,0 +1,78 @@
+import os
+
+import smtplib
+import imaplib
+import github
+
+import logging
+_logger = logging.getLogger(__name__)
+
+from .task import Task
+
+class AuthorizationCheck(Task):
+    """Task to test whether authorization is configured correctly."""
+    CONFIG = "authorization-check"
+
+    @staticmethod
+    def _validate_secret_exists(secret_name, level=logging.CRITICAL):
+        """
+        Parameters
+        ----------
+        secret_name : str
+            the name of the environment variable where the secret should be
+            stored
+        level : int
+            level at which to log message if the environment variable is
+            missing
+        """
+        secret = os.environ.get(secret_name)
+        if not secret:
+            _logger.log(
+                level,
+                f"Missing environment variable ${secret_name}.\n"
+                "Likely causes:\n"
+                " * GitHub secret of that name missing\n"
+                " * Spelling mismatch between the config name and the "
+                "GitHub secret"
+            )
+        return bool(secret)
+
+    def _run(self, config, dry):
+        _logger.debug(f"CONFIG: {config}")
+
+        # check bot token
+        valid_bot = (
+            self._validate_secret_exists(self.bot.token_secret)
+            and self.bot.validate_authorization(logger=_logger)
+        )
+
+        # check SMTP token
+        if smtp_secret_name := getattr(self.bot.smtp, 'secret', None):
+            # bot.smtp.secret is defined; check if it exists
+            valid_smtp = (
+                self._validate_secret_exists(smtp_secret_name)
+                and self.bot.smtp.validate_authorization(logger=_logger,
+                                                         label="Bot SMTP")
+            )
+        else:
+            # bot.smtp.secret not defined; warn, but call it valid because
+            # this isn't actually an error
+            valid_smtp = True
+            _logger.warn("No SMTP secret defined; sendmail not possible")
+
+        # check inbox secret
+        valid_inbox = (
+            self._validate_secret_exists(self.inbox.secret)
+            and self.inbox.validate_authorization(logger=_logger,
+                                                  label="Inbox")
+        )
+
+        if not (valid_bot and valid_smtp and valid_inbox):
+            return 1
+
+    def _build_config(self):
+        return self.config
+
+
+if __name__ == "__main__":
+    AuthorizationCheck.run_cli()

ticgithub/tasks/task.py

@@ -1,5 +1,5 @@
 import logging
-_logger = logging.getLogger("__name__")
+_logger = logging.getLogger(__name__)
 
 import yaml
 
@@ -51,7 +51,9 @@ def run_cli(cls):
         config, opts = cls.use_parser(parser)
         config, opts = cls.inject_config_parameters(config, opts)
         task = cls.from_config(config)
-        task(opts.dry)
+        result = task(opts.dry)
+        if result:
+            exit(result)
 
     @classmethod
     def from_config(cls, cfg_dict):
@@ -94,4 +96,4 @@ def __call__(self, dry=False):
                     if k not in {"active", "dry"}}
         cfg = self._build_config()
         _logger.info(f"Running {self} with config {cfg}")
-        self._run(cfg, dry)
+        return self._run(cfg, dry)

ticgithub/tests/test_bot.py

@@ -0,0 +1,55 @@
+import pytest
+
+import logging
+_logger = logging.getLogger(__name__)
+
+from ticgithub.bot import *
+
+from .utils import get_envvars_or_skip
+
+
+@pytest.fixture
+def realsmtp():
+    user, host, secret = get_envvars_or_skip(
+        "BOT_SMTP_USER",
+        "BOT_SMTP_HOST",
+        "BOT_SMTP_SECRET_NAME",
+    )
+    return SMTP(host, user, secret)
+
+
+@pytest.fixture
+def realbot(realsmtp):
+    token_secret_name, repo = get_envvars_or_skip(
+        "BOT_TOKEN_SECRET_NAME",
+        "BOT_DEMO_REPO",
+    )
+    return Bot(token_secret_name, repo, realsmtp)
+
+
+class TestSMTP:
+    @pytest.mark.parametrize("auth", [True, False])
+    def test_check_authorization(self, realsmtp, auth):
+        if auth is False:
+            realsmtp.secret = "foo"
+
+        with caplog.at_level(logging.INFO, logger=__name__):
+            result = realsmtp.validate_authorization(logger=logger,
+                                                     label="Bot SMTP")
+            assert result is auth
+            if auth:
+                expected = "Bot SMTP: Authorization succeeded"
+                assert len(caplog.records) == 1
+                assert expected in caplog.text
+            else:
+                expected_1 = "Bot SMTP: Authorization failed"
+                # expected_2 = ""
+                assert len(caplog.records) == 2
+                assert expected_1 in caplog.text
+                ... # expected_2 as well!
+
+class TestBot:
+    ...
+
+    def test_check_authorization(self, realbot):
+        ...

ticgithub/tests/utils.py

@@ -1,5 +1,25 @@
 from importlib import resources
+import os
+
+import pytest
+
 
 def datafile(filepath):
     return resources.files('ticgithub.tests.data').joinpath(filepath)
 
+
+def get_envvars_or_skip(*envvars):
+    vals = [os.environ.get(var) for var in envvars]
+    vals = []
+    missing = []
+    for var in envvars:
+        if val := os.environ.get(var):
+            vals.append(val)
+        else:
+            missing.append(var)
+
+    if missing:
+        pytest.skip("Missing environment variables: "
+                    + ", ".join(f"${var}" for var in missing))
+
+    return vals