fix notification email encrypt/decrypt implementation

CHANGES.rst

@@ -12,12 +12,20 @@ Changes
 
 Changes:
 --------
+- Add utility methods for `Job` to easily retrieve its various URLs.
+- Add ``weaver.wps_email_notify_timeout`` setting (default 10s) to avoid SMTP server deadlock on failing connection.
+- Modify the ``encrypt_email`` function to use an alternate strategy allowing ``decrypt_email`` on `Job` completed.
 - Add `CLI` ``execute`` options ``--output-public/-oP`` and ``--output-context/-oC OUTPUT_CONTEXT`` that add the
   specified ``X-WPS-Output-Context`` header to request the relevant output storage location of `Job` results.
+- Remove ``notification_email`` from ``GET /jobs`` query parameters.
+  Due to the nature of the encryption strategy, this cannot be supported anymore.
 
 Fixes:
 ------
-- No change.
+- Fix `Job` submitted with a ``notification_email`` not reversible from its encrypted value to retrieve the original
+  email on `Job` completion to send the notification (fixes `#568 <https://github.com/crim-ca/weaver/issues/568>`_).
+- Fix example Mako Template for email notification using an unavailable property ``${logs}``.
+  Instead, the new utility methods ``job.[...]_url`` should be used to retrieve relevant locations.
 
 .. _changes_4.32.0:
 

config/weaver.ini.example

@@ -148,6 +148,7 @@ weaver.wps_email_encrypt_rounds = 100000
 weaver.wps_email_notify_smtp_host =
 weaver.wps_email_notify_from_addr = [email protected]
 weaver.wps_email_notify_password = 123456
+weaver.wps_email_notify_timeout = 10
 weaver.wps_email_notify_port = 25
 weaver.wps_email_notify_ssl = true
 weaver.wps_email_notify_template_dir =

tests/test_notify.py

@@ -1,23 +1,150 @@
+import os
+import smtplib
+import tempfile
+import uuid
+
+import mock
 import pytest
 
-from weaver.notify import encrypt_email
+from weaver.datatype import Job
+from weaver.notify import decrypt_email, encrypt_email, notify_job_complete
+from weaver.status import Status
 
 
-def test_encrypt_email_valid():
+def test_encrypt_decrypt_email_valid():
     settings = {
         "weaver.wps_email_encrypt_salt": "salty-email",
     }
-    email = encrypt_email("[email protected]", settings)
-    assert email == "a1724b030d999322e2ecc658453f992472c63867cd3cef3b3d829d745bd80f34"
+    email = "[email protected]"
+    token = encrypt_email(email, settings)
+    assert token != email
+    value = decrypt_email(token, settings)
+    assert value == email
+
+
+def test_encrypt_email_random():
+    email = "[email protected]"
+    settings = {"weaver.wps_email_encrypt_salt": "salty-email"}
+    token1 = encrypt_email(email, settings)
+    token2 = encrypt_email(email, settings)
+    token3 = encrypt_email(email, settings)
+    assert token1 != token2 != token3
+
+    # although encrypted are all different, they should all decrypt back to the original!
+    email1 = decrypt_email(token1, settings)
+    email2 = decrypt_email(token2, settings)
+    email3 = decrypt_email(token3, settings)
+    assert email1 == email2 == email3 == email
 
 
-def test_encrypt_email_raise():
+@pytest.mark.parametrize("email_func", [encrypt_email, decrypt_email])
+def test_encrypt_decrypt_email_raise(email_func):
     with pytest.raises(TypeError):
-        encrypt_email("", {})
+        email_func("", {})
         pytest.fail("Should have raised for empty email")
     with pytest.raises(TypeError):
-        encrypt_email(1, {})
+        email_func(1, {})  # type: ignore
         pytest.fail("Should have raised for wrong type")
     with pytest.raises(ValueError):
-        encrypt_email("[email protected]", {})
+        email_func("[email protected]", {})
         pytest.fail("Should have raised for invalid/missing settings")
+
+
+def test_notify_job_complete():
+    test_url = "https://test-weaver.example.com"
+    settings = {
+        "weaver.url": test_url,
+        "weaver.wps_email_notify_smtp_host": "xyz.test.com",
+        "weaver.wps_email_notify_from_addr": "[email protected]",
+        "weaver.wps_email_notify_password": "super-secret",
+        "weaver.wps_email_notify_port": 12345,
+        "weaver.wps_email_notify_timeout": 1,  # quick fail if invalid
+    }
+    notify_email = "[email protected]"
+    test_job = Job(
+        task_id=uuid.uuid4(),
+        process="test-process",
+        settings=settings,
+    )
+    test_job_err_url = f"{test_url}/processes/{test_job.process}/jobs/{test_job.id}/exceptions"
+    test_job_out_url = f"{test_url}/processes/{test_job.process}/jobs/{test_job.id}/results"
+    test_job_log_url = f"{test_url}/processes/{test_job.process}/jobs/{test_job.id}/logs"
+
+    with mock.patch("smtplib.SMTP_SSL", autospec=smtplib.SMTP_SSL) as mock_smtp:
+        mock_smtp.return_value.sendmail.return_value = None  # sending worked
+
+        test_job.status = Status.SUCCEEDED
+        notify_job_complete(test_job, notify_email, settings)
+        mock_smtp.assert_called_with("xyz.test.com", 12345, timeout=1)
+        assert mock_smtp.return_value.sendmail.call_args[0][0] == "[email protected]"
+        assert mock_smtp.return_value.sendmail.call_args[0][1] == notify_email
+        message_encoded = mock_smtp.return_value.sendmail.call_args[0][2]
+        assert message_encoded
+        message = message_encoded.decode("utf8")
+        assert "From: Weaver" in message
+        assert f"To: {notify_email}" in message
+        assert f"Subject: Job {test_job.process} Succeeded"
+        assert test_job_out_url in message
+        assert test_job_log_url in message
+        assert test_job_err_url not in message
+
+        test_job.status = Status.FAILED
+        notify_job_complete(test_job, notify_email, settings)
+        assert mock_smtp.return_value.sendmail.call_args[0][0] == "[email protected]"
+        assert mock_smtp.return_value.sendmail.call_args[0][1] == notify_email
+        message_encoded = mock_smtp.return_value.sendmail.call_args[0][2]
+        assert message_encoded
+        message = message_encoded.decode("utf8")
+        assert "From: Weaver" in message
+        assert f"To: {notify_email}" in message
+        assert f"Subject: Job {test_job.process} Failed"
+        assert test_job_out_url not in message
+        assert test_job_log_url in message
+        assert test_job_err_url in message
+
+
+def test_notify_job_complete_custom_template():
+    with tempfile.NamedTemporaryFile(mode="w", encoding="utf-8", suffix=".mako") as email_template_file:
+        email_template_file.writelines([
+            "From: Weaver\n",
+            "To: ${to}\n",
+            "Subject: Job ${job.process} ${job.status}\n",
+            "\n",  # end of email header, content below
+            "Job: ${job.status_url(settings)}\n",
+        ])
+        email_template_file.flush()
+        email_template_file.seek(0)
+
+        mako_dir, mako_name = os.path.split(email_template_file.name)
+        test_url = "https://test-weaver.example.com"
+        settings = {
+            "weaver.url": test_url,
+            "weaver.wps_email_notify_smtp_host": "xyz.test.com",
+            "weaver.wps_email_notify_from_addr": "[email protected]",
+            "weaver.wps_email_notify_password": "super-secret",
+            "weaver.wps_email_notify_port": 12345,
+            "weaver.wps_email_notify_timeout": 1,  # quick fail if invalid
+            "weaver.wps_email_notify_template_dir": mako_dir,
+            "weaver.wps_email_notify_template_default": mako_name,
+        }
+        notify_email = "[email protected]"
+        test_job = Job(
+            task_id=uuid.uuid4(),
+            process="test-process",
+            status=Status.SUCCEEDED,
+            settings=settings,
+        )
+
+        with mock.patch("smtplib.SMTP_SSL", autospec=smtplib.SMTP_SSL) as mock_smtp:
+            mock_smtp.return_value.sendmail.return_value = None  # sending worked
+            notify_job_complete(test_job, notify_email, settings)
+
+        message_encoded = mock_smtp.return_value.sendmail.call_args[0][2]
+        message = message_encoded.decode("utf8")
+        assert message == "\n".join([
+            "From: Weaver",
+            f"To: {notify_email}",
+            f"Subject: Job {test_job.process} {Status.SUCCEEDED}",
+            "",
+            f"Job: {test_url}/processes/{test_job.process}/jobs/{test_job.id}",
+        ])

tests/wps_restapi/test_jobs.py

@@ -1,4 +1,5 @@
 import contextlib
+import copy
 import datetime
 import logging
 import os
@@ -34,6 +35,7 @@
 from weaver.datatype import Job, Service
 from weaver.execute import ExecuteMode, ExecuteResponse, ExecuteTransmissionMode
 from weaver.formats import ContentType
+from weaver.notify import decrypt_email
 from weaver.processes.wps_testing import WpsTestProcess
 from weaver.status import JOB_STATUS_CATEGORIES, Status, StatusCategory
 from weaver.utils import get_path_kvp, now
@@ -543,6 +545,7 @@ def test_get_jobs_page_out_of_range(self):
         assert "limit" in str(resp.json["cause"]) and "less than minimum" in str(resp.json["cause"])
         assert "limit" in resp.json["value"] and resp.json["value"]["limit"] == str(0)
 
+    @pytest.mark.skip(reason="Obsolete feature. It is not possible to filter by encrypted notification email anymore.")
     def test_get_jobs_by_encrypted_email(self):
         """
         Verifies that literal email can be used as search criterion although not saved in plain text within db.
@@ -562,13 +565,20 @@ def test_get_jobs_by_encrypted_email(self):
             resp = self.app.post_json(path, params=body, headers=self.json_headers)
             assert resp.status_code == 201
             assert resp.content_type == ContentType.APP_JSON
-        job_id = resp.json["jobID"]
+            job_id = resp.json["jobID"]
+
+            # submit a second job just to make sure email doesn't match it as well
+            other_body = copy.deepcopy(body)
+            other_body["notification_email"] = "[email protected]"
+            resp = self.app.post_json(path, params=other_body, headers=self.json_headers)
+            assert resp.status_code == 201
 
         # verify the email is not in plain text
         job = self.job_store.fetch_by_id(job_id)
         assert job.notification_email != email and job.notification_email is not None
-        assert int(job.notification_email, 16) != 0  # email should be encrypted with hex string
+        assert decrypt_email(job.notification_email, self.settings) == email, "Email should be recoverable."
 
+        # make sure that jobs searched using email are found with encryption transparently for the user
         path = get_path_kvp(sd.jobs_service.path, detail="true", notification_email=email)
         resp = self.app.get(path, headers=self.json_headers)
         assert resp.status_code == 200

weaver/datatype.py

@@ -925,17 +925,6 @@
             raise TypeError(f"Type 'str' is required for '{self.__name__}.status_location'")
         self["status_location"] = location_url
 
-    def status_url(self, container=None):
-        # type: (Optional[AnySettingsContainer]) -> str
-        """
-        Obtain the resolved endpoint where the :term:`Job` status information can be obtained.
-        """
-        settings = get_settings(container)
-        location_base = f"/providers/{self.service}" if self.service else ""
-        api_base_url = get_wps_restapi_base_url(settings)
-        location_url = f"{api_base_url}{location_base}/processes/{self.process}/jobs/{self.id}"
-        return location_url
-
     @property
     def notification_email(self):
         # type: () -> Optional[str]
@@ -1236,13 +1225,43 @@
             response = xml_util.tostring(response)
         self["response"] = response
 
-    def _job_url(self, base_url=None):
-        # type: (Optional[str]) -> str
+    def _job_url(self, base_url):
+        # type: (str) -> str
         if self.service is not None:
             base_url += sd.provider_service.path.format(provider_id=self.service)
         job_path = sd.process_job_service.path.format(process_id=self.process, job_id=self.id)
         return base_url + job_path
 
+    def job_url(self, container=None, extra_path=None):
+        # type: (Optional[AnySettingsContainer], Optional[str]) -> str
+        settings = get_settings(container)
+        base_url = get_wps_restapi_base_url(settings)
+        return self._job_url(base_url) + (extra_path or "")
+
+    def status_url(self, container=None):
+        # type: (Optional[AnySettingsContainer]) -> str
+        return self.job_url(container=container)
+
+    def logs_url(self, container=None):
+        # type: (Optional[AnySettingsContainer]) -> str
+        return self.job_url(container=container, extra_path="/logs")
+
+    def exceptions_url(self, container=None):
+        # type: (Optional[AnySettingsContainer]) -> str
+        return self.job_url(container=container, extra_path="/exceptions")
+
+    def inputs_url(self, container=None):
+        # type: (Optional[AnySettingsContainer]) -> str
+        return self.job_url(container=container, extra_path="/inputs")
+
+    def outputs_url(self, container=None):
+        # type: (Optional[AnySettingsContainer]) -> str
+        return self.job_url(container=container, extra_path="/outputs")
+
+    def results_url(self, container=None):
+        # type: (Optional[AnySettingsContainer]) -> str
+        return self.job_url(container=container, extra_path="/results")
+
     def links(self, container=None, self_link=None):
         # type: (Optional[AnySettingsContainer], Optional[str]) -> List[Link]
         """