from datetime import datetime, UTC from freezegun import freeze_time import pytest MAIL_STRING = rb"""Delivered-To: to@test1.com MIME-Version: 1.0 From: John Smith Date: Mon, 10 Aug 2020 10:17:16 +0300 Subject: Testing email for mail listener To: to@test1.com Content-Type: multipart/alternative; boundary="0000000000002b271405ac80bf8b" --0000000000002b271405ac80bf8b Content-Type: text/plain; charset="UTF-8" --0000000000002b271405ac80bf8b Content-Type: text/html; charset="UTF-8"

C:\Users

C:\\Users

--0000000000002b271405ac80bf8b-- """ MAIL_STRING_NO_DATE = rb"""Delivered-To: to@test1.com MIME-Version: 1.0 From: John Smith Date: Subject: Testing email for mail listener To: to@test1.com Content-Type: multipart/alternative; boundary="0000000000002b271405ac80bf8b" --0000000000002b271405ac80bf8b Content-Type: text/plain; charset="UTF-8" --0000000000002b271405ac80bf8b Content-Type: text/html; charset="UTF-8"

C:\Users

C:\\Users

--0000000000002b271405ac80bf8b-- """ MAIL_STRING_NOT_BYTES = r"""Delivered-To: to@test1.com MIME-Version: 1.0 From: John Smith Date: Mon, 10 Aug 2020 10:17:16 +0300 Subject: Testing email for mail listener To: to@test1.com Content-Type: multipart/alternative; boundary="0000000000002b271405ac80bf8b" --0000000000002b271405ac80bf8b Content-Type: text/plain; charset="UTF-8" --0000000000002b271405ac80bf8b Content-Type: text/html; charset="UTF-8"

C:\Users

--0000000000002b271405ac80bf8b-- """ EXPECTED_LABELS = [ {"type": "Email/from", "value": "from@test1.com"}, {"type": "Email/format", "value": "multipart/alternative"}, {"type": "Email/text", "value": ""}, {"type": "Email/subject", "value": "Testing email for mail listener"}, {"type": "Email/headers/Delivered-To", "value": "to@test1.com"}, {"type": "Email/headers/MIME-Version", "value": "1.0"}, {"type": "Email/headers/From", "value": "John Smith "}, {"type": "Email/headers/Date", "value": "Mon, 10 Aug 2020 10:17:16 +0300"}, {"type": "Email/headers/Subject", "value": "Testing email for mail listener"}, {"type": "Email/headers/To", "value": "to@test1.com"}, {"type": "Email/headers/Content-Type", "value": 'multipart/alternative; boundary="0000000000002b271405ac80bf8b"'}, {"type": "Email", "value": "to@test1.com"}, {"type": "Email/html", "value": '

\n

C:\\Users

\n

C:\\\\Users

'}, ] @freeze_time("2022-12-11 13:40:00 UTC") @pytest.mark.parametrize( "mail_string, mail_date", [(MAIL_STRING, "2020-08-10T07:17:16+00:00"), (MAIL_STRING_NO_DATE, "2022-12-11T13:40:00+00:00")] ) def test_convert_to_incident(mail_string, mail_date): """ Given: - Bytes representation of a mail When: - Parsing it to incidents Then: - Validate the 'attachments', 'occurred', 'details' and 'name' fields are parsed as expected """ from MailListenerV2 import Email email = Email(mail_string, False, False, 0) incident = email.convert_to_incident() assert incident["occurred"] == mail_date assert incident["details"] == email.text or email.html assert incident["name"] == email.subject @pytest.mark.parametrize( "time_to_fetch_from, with_header, permitted_from_addresses, permitted_from_domains, uid_to_fetch_from, expected_query", # noqa: E501 [ ( datetime(year=2020, month=10, day=1), # noqa: E126 False, # noqa: E126 ["test1@mail.com", "test2@mail.com"], ["test1.com", "domain2.com"], 4, [ "OR", "OR", "OR", "FROM", "domain2.com", "FROM", "test1.com", "FROM", "test1@mail.com", "FROM", "test2@mail.com", "SINCE", datetime(year=2020, month=10, day=1), "UID", "4:*", ], ), ( datetime(year=2020, month=10, day=1), # noqa: E126 True, # noqa: E126 ["test1@mail.com", "test2@mail.com"], ["test1.com", "domain2.com"], 4, [ "OR", "OR", "OR", "HEADER", "FROM", "domain2.com", "HEADER", "FROM", "test1.com", "HEADER", "FROM", "test1@mail.com", "HEADER", "FROM", "test2@mail.com", "SINCE", datetime(year=2020, month=10, day=1), "UID", "4:*", ], ), ( None, # noqa: E126 "", # noqa: E126 [], [], 1, ["UID", "1:*"], ), ], ) def test_generate_search_query( time_to_fetch_from, with_header, permitted_from_addresses, permitted_from_domains, uid_to_fetch_from, expected_query ): """ Given: - The date from which mails should be queried - A list of email addresses from which mails should be queried - A list of domains from which mails should be queried When: - Generating search query from these arguments Then: - Validate the search query as enough 'OR's in the beginning (Σ(from n=0to(len(addresses)+len(domains)))s^(n-1)) - Validate the search query has FROM before each address or domain - Validate query has SINCE before the datetime object """ from MailListenerV2 import generate_search_query assert ( generate_search_query( time_to_fetch_from, with_header, permitted_from_addresses, permitted_from_domains, uid_to_fetch_from ) == expected_query ) def test_generate_labels(): """ Given: - Bytes representation of a mail When: - Generating mail labels Then: - Validate all expected labels are in the generated labels, and are from type string """ from MailListenerV2 import Email email = Email(MAIL_STRING, False, False, 0) labels = email._generate_labels() for label in EXPECTED_LABELS: assert label in labels, f"Label {label} was not found in the generated labels, {labels}" def mock_email(): from unittest.mock import patch from MailListenerV2 import Email with patch.object(Email, "__init__", lambda a, b, c, d, e: None): email = Email("data", False, False, 0) email.id = 0 email.date = datetime(year=2020, month=10, day=1, tzinfo=UTC) return email def mock_email_without_date(mail_id=0): """Build a mock Email whose Date header could not be parsed, so its date is None.""" from unittest.mock import patch from MailListenerV2 import Email with patch.object(Email, "__init__", lambda a, b, c, d, e: None): email = Email("data", False, False, 0) email.id = mail_id email.date = None return email @pytest.mark.parametrize( "src_data, expected", [({1: {b"RFC822": rb"C:\User\u"}}, rb"C:\User\u"), ({2: {b"RFC822": rb"C:\User\u"}}, rb"C:\User\u")] ) def test_fetch_mail_gets_bytes(mocker, src_data, expected): """ Given: A byte string representing response from API When: 1. The string returns as string 2. The string returns as bytes Then: validates Email is called with a bytes string. """ from MailListenerV2 import fetch_mails import demistomock as demisto from imapclient import IMAPClient mail_mocker = mocker.patch("MailListenerV2.Email", return_value=mock_email()) mocker.patch.object(demisto, "debug") mocker.patch.object(IMAPClient, "search", return_value=[1]) mocker.patch.object(IMAPClient, "fetch", return_value=src_data) mocker.patch.object(IMAPClient, "_create_IMAP4") fetch_mails(IMAPClient("http://example_url.com")) assert mail_mocker.call_args[0][0] == expected def test_fetch_mail__default_uid( mocker, ): """ Given: - No uid is passed to fetch_mails function When: - Fetching mails Then: - Validate search query is empty """ from MailListenerV2 import fetch_mails import demistomock as demisto from imapclient import IMAPClient mocker.patch("MailListenerV2.Email") mocker.patch.object(demisto, "debug") search_mocker = mocker.patch.object(IMAPClient, "search", return_value=[1]) mocker.patch.object(IMAPClient, "fetch") mocker.patch.object(IMAPClient, "_create_IMAP4") fetch_mails(IMAPClient("http://example_url.com")) assert search_mocker.call_args[0][0] == [] # default uid is 0 so no search query is passed def test_invalid_mail_object_handling(mocker): """ Given: - Fetch response with 3 mails, the 2nd being invalid mail When: - Fetching mails Then: - Validate only 2 valid mails are returned - Validate skipping invalid mail and printing relevant debug message """ src_data = {1: {b"RFC822": rb"C:\User1\u"}, 2: {b"RFC822": rb"C:\User2\u"}, 3: {b"RFC822": rb"C:\User3\u"}} from MailListenerV2 import fetch_mails import demistomock as demisto from imapclient import IMAPClient mock_email_1 = mock_email() mock_email_3 = mock_email() mock_email_1.id, mock_email_3.id = 10, 11 mocker.patch("MailListenerV2.Email", side_effect=[mock_email_1, Exception("Invalid Mail"), mock_email_3]) mocker.patch.object(demisto, "debug") mocker.patch.object(IMAPClient, "search", return_value=[1]) mocker.patch.object(IMAPClient, "fetch", return_value=src_data) mocker.patch.object(IMAPClient, "_create_IMAP4") mails_fetched, messages_fetched, _ = fetch_mails(IMAPClient("http://example_url.com")) assert len(mails_fetched) == 2 assert messages_fetched == [10, 11] def test_get_eml_attachments(): from MailListenerV2 import Email import email # Test an email with a PNG attachment with open("test_data/eml-with-jpeg.eml", "rb") as f: msg = email.message_from_bytes(f.read()) res = Email.get_eml_attachments(msg.as_bytes()) assert res == [] # Test an email with EML attachment with open("test_data/eml-with-eml-with-attachment.eml", "rb") as f: msg = email.message_from_bytes(f.read()) res = Email.get_eml_attachments(msg.as_bytes()) assert res[0]["filename"] == "Test with an image.eml" # Test an email with EML attachment with an EML attachment with open("test_data/eml_test_with_attachment_with_eml_attachment.eml", "rb") as f: msg = email.message_from_bytes(f.read()) res = Email.get_eml_attachments(msg.as_bytes()) assert res[0]["filename"] == "Fwd: MOIS DE MARSè.eml" assert isinstance(res[0]["payload"], bytes) @pytest.mark.parametrize( "cert_and_key", [ # - cert and key are in the integration instance parameters # - private key is OpenSSL format # *** The cert and key below are not used in the real services, and only used for testing. ( { "password": "-----BEGIN CERTIFICATE----- " "MIIDlzCCAX+gAwIBAgIUbN3atZY05K7SilRtY78y2ZON28QwDQYJKoZIhvcNAQEN " "BQAwJTEjMCEGA1UEAwwaTWFpbCBMaXN0ZW5lciBUZXN0IFJvb3QgQ0EwHhcNMjMw " "NzExMDA1NzI4WhcNMzMwNzExMDA1NzI4WjAaMRgwFgYDVQQDDA90ZXN0IGNsaWVu " "dCBlY2MwdjAQBgcqhkjOPQIBBgUrgQQAIgNiAARFTRK4qjfOkK25NAssTni1/bKD " "TvEtmBFy5N0Qi+kisnUS05e9Okp2d8txhClwbjFbiunaNcHxdKJkOY6p/VFpzERE " "gtGiLBVZf8YrYOBPHc93tFiWs7+z1C63uNRUVGujeDB2MB8GA1UdIwQYMBaAFDh6 " "N1cbIXsS4uo15Ha9fKZrEbcHMB0GA1UdDgQWBBRFSQMsVOPCmzMbvjnrYMGF1ZNs " "8DAMBgNVHRMBAf8EAjAAMA4GA1UdDwEB/wQEAwIE8DAWBgNVHSUBAf8EDDAKBggr " "BgEFBQcDAjANBgkqhkiG9w0BAQ0FAAOCAgEAJKmlIV9Du9pnA98vw4GsurAeXU3Y " "KlzMyffzIVF+CGTpUFmXIGu1KccZREGQEZpxotYF71HCCqPBUcQD8rRoetxX3wa6 " "iqk6Q3Pm9Jt8/P365vydvcsKvTEeP8NTWKVip7U8xgAIjykBdnEPu9Uq7x+bePiG " "Pqd2Mpzr+mydbU/3mzrZXm/3B0aNiYZdXSpkF4qwZ7lakFvn0MI1M9+B2Am+rNdJ " "AoBBQTwS+1pUZoKV3gXMRWKCHj5cbltf1+Lzhh64A8s8k0o1cFyXfSFZ/PJI2rve " "ZOKGQ8qIeF3FPCaX5TVvla9J5Mxz5ETXv5zWpK/H4VgPbLf1cZPGFHnYatKkXMvM " "05UZ1FmdmJSS8CQQ7AwRsAyWOrbnfUf3Xv5UVFlgGYbsM1+ENbs9Mpn9mq0zq7+J " "ONxkmyrkP3Gi/ZK1k9fZuE+WGrnzkP6zUMA76Zr2uH8Gq5Bt89jTl9gAAYuaIDSe " "TQDQuO+Pb6XYiJaUg3LbkAnUSQHawZ6DfAMghevCPTIrTFLTUi8gILIpN2ghfv+z " "R2DE2xaKvNzNgEfPxR94haUGZy6eExteWFVbJAbQotux2poksrFqdgTW/7qntrpN " "l2AxOYvV/yu0yDjf/kyzt2aoWsbxClNv3jrbAj3m6raY/e6lcr6IuMYWMtO2F3n+ " "OzZEXmZyHr121wY= " "-----END CERTIFICATE----- " "-----BEGIN EC PRIVATE KEY----- " "MIGkAgEBBDCUBWVfn8bslTSkoWyA47lB8CwM/R5dlHrH4R52FkCmFFttnlotCt2v " "OCzaIX4lCIygBwYFK4EEACKhZANiAARFTRK4qjfOkK25NAssTni1/bKDTvEtmBFy " "5N0Qi+kisnUS05e9Okp2d8txhClwbjFbiunaNcHxdKJkOY6p/VFpzEREgtGiLBVZ " "f8YrYOBPHc93tFiWs7+z1C63uNRUVGs= " "-----END EC PRIVATE KEY-----" } ), # - cert and key are in the Certificate section of the Credentials # - private key is OpenSSL format # *** The cert and key below are not used in the real services, and only used for testing. ( { "credentials": { "sshkey": """ -----BEGIN CERTIFICATE----- MIIDlzCCAX+gAwIBAgIUbN3atZY05K7SilRtY78y2ZON28QwDQYJKoZIhvcNAQEN BQAwJTEjMCEGA1UEAwwaTWFpbCBMaXN0ZW5lciBUZXN0IFJvb3QgQ0EwHhcNMjMw NzExMDA1NzI4WhcNMzMwNzExMDA1NzI4WjAaMRgwFgYDVQQDDA90ZXN0IGNsaWVu dCBlY2MwdjAQBgcqhkjOPQIBBgUrgQQAIgNiAARFTRK4qjfOkK25NAssTni1/bKD TvEtmBFy5N0Qi+kisnUS05e9Okp2d8txhClwbjFbiunaNcHxdKJkOY6p/VFpzERE gtGiLBVZf8YrYOBPHc93tFiWs7+z1C63uNRUVGujeDB2MB8GA1UdIwQYMBaAFDh6 N1cbIXsS4uo15Ha9fKZrEbcHMB0GA1UdDgQWBBRFSQMsVOPCmzMbvjnrYMGF1ZNs 8DAMBgNVHRMBAf8EAjAAMA4GA1UdDwEB/wQEAwIE8DAWBgNVHSUBAf8EDDAKBggr BgEFBQcDAjANBgkqhkiG9w0BAQ0FAAOCAgEAJKmlIV9Du9pnA98vw4GsurAeXU3Y KlzMyffzIVF+CGTpUFmXIGu1KccZREGQEZpxotYF71HCCqPBUcQD8rRoetxX3wa6 iqk6Q3Pm9Jt8/P365vydvcsKvTEeP8NTWKVip7U8xgAIjykBdnEPu9Uq7x+bePiG Pqd2Mpzr+mydbU/3mzrZXm/3B0aNiYZdXSpkF4qwZ7lakFvn0MI1M9+B2Am+rNdJ AoBBQTwS+1pUZoKV3gXMRWKCHj5cbltf1+Lzhh64A8s8k0o1cFyXfSFZ/PJI2rve ZOKGQ8qIeF3FPCaX5TVvla9J5Mxz5ETXv5zWpK/H4VgPbLf1cZPGFHnYatKkXMvM 05UZ1FmdmJSS8CQQ7AwRsAyWOrbnfUf3Xv5UVFlgGYbsM1+ENbs9Mpn9mq0zq7+J ONxkmyrkP3Gi/ZK1k9fZuE+WGrnzkP6zUMA76Zr2uH8Gq5Bt89jTl9gAAYuaIDSe TQDQuO+Pb6XYiJaUg3LbkAnUSQHawZ6DfAMghevCPTIrTFLTUi8gILIpN2ghfv+z R2DE2xaKvNzNgEfPxR94haUGZy6eExteWFVbJAbQotux2poksrFqdgTW/7qntrpN l2AxOYvV/yu0yDjf/kyzt2aoWsbxClNv3jrbAj3m6raY/e6lcr6IuMYWMtO2F3n+ OzZEXmZyHr121wY= -----END CERTIFICATE----- -----BEGIN EC PRIVATE KEY----- MIGkAgEBBDCUBWVfn8bslTSkoWyA47lB8CwM/R5dlHrH4R52FkCmFFttnlotCt2v OCzaIX4lCIygBwYFK4EEACKhZANiAARFTRK4qjfOkK25NAssTni1/bKDTvEtmBFy 5N0Qi+kisnUS05e9Okp2d8txhClwbjFbiunaNcHxdKJkOY6p/VFpzEREgtGiLBVZ f8YrYOBPHc93tFiWs7+z1C63uNRUVGs= -----END EC PRIVATE KEY----- """ } } ), # - cert and key are in the integration instance parameters # - private key is PKCS#8 PEM # *** The cert and key below are not used in the real services, and only used for testing. ( { "password": "-----BEGIN CERTIFICATE----- " "MIIDlzCCAX+gAwIBAgIUbN3atZY05K7SilRtY78y2ZON28QwDQYJKoZIhvcNAQEN " "BQAwJTEjMCEGA1UEAwwaTWFpbCBMaXN0ZW5lciBUZXN0IFJvb3QgQ0EwHhcNMjMw " "NzExMDA1NzI4WhcNMzMwNzExMDA1NzI4WjAaMRgwFgYDVQQDDA90ZXN0IGNsaWVu " "dCBlY2MwdjAQBgcqhkjOPQIBBgUrgQQAIgNiAARFTRK4qjfOkK25NAssTni1/bKD " "TvEtmBFy5N0Qi+kisnUS05e9Okp2d8txhClwbjFbiunaNcHxdKJkOY6p/VFpzERE " "gtGiLBVZf8YrYOBPHc93tFiWs7+z1C63uNRUVGujeDB2MB8GA1UdIwQYMBaAFDh6 " "N1cbIXsS4uo15Ha9fKZrEbcHMB0GA1UdDgQWBBRFSQMsVOPCmzMbvjnrYMGF1ZNs " "8DAMBgNVHRMBAf8EAjAAMA4GA1UdDwEB/wQEAwIE8DAWBgNVHSUBAf8EDDAKBggr " "BgEFBQcDAjANBgkqhkiG9w0BAQ0FAAOCAgEAJKmlIV9Du9pnA98vw4GsurAeXU3Y " "KlzMyffzIVF+CGTpUFmXIGu1KccZREGQEZpxotYF71HCCqPBUcQD8rRoetxX3wa6 " "iqk6Q3Pm9Jt8/P365vydvcsKvTEeP8NTWKVip7U8xgAIjykBdnEPu9Uq7x+bePiG " "Pqd2Mpzr+mydbU/3mzrZXm/3B0aNiYZdXSpkF4qwZ7lakFvn0MI1M9+B2Am+rNdJ " "AoBBQTwS+1pUZoKV3gXMRWKCHj5cbltf1+Lzhh64A8s8k0o1cFyXfSFZ/PJI2rve " "ZOKGQ8qIeF3FPCaX5TVvla9J5Mxz5ETXv5zWpK/H4VgPbLf1cZPGFHnYatKkXMvM " "05UZ1FmdmJSS8CQQ7AwRsAyWOrbnfUf3Xv5UVFlgGYbsM1+ENbs9Mpn9mq0zq7+J " "ONxkmyrkP3Gi/ZK1k9fZuE+WGrnzkP6zUMA76Zr2uH8Gq5Bt89jTl9gAAYuaIDSe " "TQDQuO+Pb6XYiJaUg3LbkAnUSQHawZ6DfAMghevCPTIrTFLTUi8gILIpN2ghfv+z " "R2DE2xaKvNzNgEfPxR94haUGZy6eExteWFVbJAbQotux2poksrFqdgTW/7qntrpN " "l2AxOYvV/yu0yDjf/kyzt2aoWsbxClNv3jrbAj3m6raY/e6lcr6IuMYWMtO2F3n+ " "OzZEXmZyHr121wY= " "-----END CERTIFICATE----- " "-----BEGIN PRIVATE KEY----- " "MIG2AgEAMBAGByqGSM49AgEGBSuBBAAiBIGeMIGbAgEBBDCUBWVfn8bslTSkoWyA " "47lB8CwM/R5dlHrH4R52FkCmFFttnlotCt2vOCzaIX4lCIyhZANiAARFTRK4qjfO " "kK25NAssTni1/bKDTvEtmBFy5N0Qi+kisnUS05e9Okp2d8txhClwbjFbiunaNcHx " "dKJkOY6p/VFpzEREgtGiLBVZf8YrYOBPHc93tFiWs7+z1C63uNRUVGs= " "-----END PRIVATE KEY-----" } ), # - cert and key are in the Certificate secion of the Credentials # - private key is PKCS#8 PEM # *** The cert and key below are not used in the real services, and only used for testing. ( { "credentials": { "sshkey": """ -----BEGIN CERTIFICATE----- MIIDlzCCAX+gAwIBAgIUbN3atZY05K7SilRtY78y2ZON28QwDQYJKoZIhvcNAQEN BQAwJTEjMCEGA1UEAwwaTWFpbCBMaXN0ZW5lciBUZXN0IFJvb3QgQ0EwHhcNMjMw NzExMDA1NzI4WhcNMzMwNzExMDA1NzI4WjAaMRgwFgYDVQQDDA90ZXN0IGNsaWVu dCBlY2MwdjAQBgcqhkjOPQIBBgUrgQQAIgNiAARFTRK4qjfOkK25NAssTni1/bKD TvEtmBFy5N0Qi+kisnUS05e9Okp2d8txhClwbjFbiunaNcHxdKJkOY6p/VFpzERE gtGiLBVZf8YrYOBPHc93tFiWs7+z1C63uNRUVGujeDB2MB8GA1UdIwQYMBaAFDh6 N1cbIXsS4uo15Ha9fKZrEbcHMB0GA1UdDgQWBBRFSQMsVOPCmzMbvjnrYMGF1ZNs 8DAMBgNVHRMBAf8EAjAAMA4GA1UdDwEB/wQEAwIE8DAWBgNVHSUBAf8EDDAKBggr BgEFBQcDAjANBgkqhkiG9w0BAQ0FAAOCAgEAJKmlIV9Du9pnA98vw4GsurAeXU3Y KlzMyffzIVF+CGTpUFmXIGu1KccZREGQEZpxotYF71HCCqPBUcQD8rRoetxX3wa6 iqk6Q3Pm9Jt8/P365vydvcsKvTEeP8NTWKVip7U8xgAIjykBdnEPu9Uq7x+bePiG Pqd2Mpzr+mydbU/3mzrZXm/3B0aNiYZdXSpkF4qwZ7lakFvn0MI1M9+B2Am+rNdJ AoBBQTwS+1pUZoKV3gXMRWKCHj5cbltf1+Lzhh64A8s8k0o1cFyXfSFZ/PJI2rve ZOKGQ8qIeF3FPCaX5TVvla9J5Mxz5ETXv5zWpK/H4VgPbLf1cZPGFHnYatKkXMvM 05UZ1FmdmJSS8CQQ7AwRsAyWOrbnfUf3Xv5UVFlgGYbsM1+ENbs9Mpn9mq0zq7+J ONxkmyrkP3Gi/ZK1k9fZuE+WGrnzkP6zUMA76Zr2uH8Gq5Bt89jTl9gAAYuaIDSe TQDQuO+Pb6XYiJaUg3LbkAnUSQHawZ6DfAMghevCPTIrTFLTUi8gILIpN2ghfv+z R2DE2xaKvNzNgEfPxR94haUGZy6eExteWFVbJAbQotux2poksrFqdgTW/7qntrpN l2AxOYvV/yu0yDjf/kyzt2aoWsbxClNv3jrbAj3m6raY/e6lcr6IuMYWMtO2F3n+ OzZEXmZyHr121wY= -----END CERTIFICATE----- -----BEGIN PRIVATE KEY----- MIG2AgEAMBAGByqGSM49AgEGBSuBBAAiBIGeMIGbAgEBBDCUBWVfn8bslTSkoWyA 47lB8CwM/R5dlHrH4R52FkCmFFttnlotCt2vOCzaIX4lCIyhZANiAARFTRK4qjfO kK25NAssTni1/bKDTvEtmBFy5N0Qi+kisnUS05e9Okp2d8txhClwbjFbiunaNcHx dKJkOY6p/VFpzEREgtGiLBVZf8YrYOBPHc93tFiWs7+z1C63uNRUVGs= -----END PRIVATE KEY----- """ } } ), ], ) def test_load_client_cert_and_key(mocker, cert_and_key): """ Given: Client cetifcates and private keys from the integration's parameters When: Authenticating the client using SSL client certificate authentication Then: 1. Validate that the SSLContext object, that is used for authentication, is given the correct file that holds the certificates 2. The function 'load_client_cert_and_key' returns True, inidicating that we are using SSL client certificate authentication """ from MailListenerV2 import load_client_cert_and_key import ssl import tempfile params = {"clientCertAndKey": cert_and_key} named_temporary_file_mocker = mocker.patch( "MailListenerV2.NamedTemporaryFile", return_value=tempfile.NamedTemporaryFile(mode="w") ) ssl_ctx = ssl.create_default_context() ssl_ctx.check_hostname = False ssl_ctx.verify_mode = ssl.CERT_NONE ssl_ctx.minimum_version = ssl.TLSVersion.TLSv1_2 load_cert_chain_mocker = mocker.patch.object(ssl_ctx, "load_cert_chain") assert load_client_cert_and_key(ssl_ctx, params) is True assert load_cert_chain_mocker.call_args_list[0][1].get("certfile") == named_temporary_file_mocker.return_value.name def test_load_empty_client_cert_and_key_(): """ Given: Client cetifcates and private keys are not configured in the integration's parameters When: Authenticating the client Then: The function 'load_client_cert_and_key' returns False, inidicating that we are not using SSL client certificate authentication """ from MailListenerV2 import load_client_cert_and_key import ssl # - No certificates and private keys params: dict[str, dict] = {"clientCertAndKey": {}} ssl_ctx = ssl.create_default_context() ssl_ctx.check_hostname = False ssl_ctx.verify_mode = ssl.CERT_NONE ssl_ctx.minimum_version = ssl.TLSVersion.TLSv1_2 assert load_client_cert_and_key(ssl_ctx, params) is False @pytest.mark.parametrize( "input_credentials, output_credentials", [ # No credentials (None, None), # 1 credential ( "-----BEGIN CERTIFICATE----- LINE1 LINE2 -----END CERTIFICATE-----", """-----BEGIN CERTIFICATE----- LINE1 LINE2 -----END CERTIFICATE-----""", ), # 2 credentials ( "-----BEGIN CERTIFICATE----- " "LINE1 " "LINE2 " "-----END CERTIFICATE----- " "-----BEGIN EC PRIVATE KEY----- " "LINE1 " "LINE2 " "-----END EC PRIVATE KEY-----", """-----BEGIN CERTIFICATE----- LINE1 LINE2 -----END CERTIFICATE----- -----BEGIN EC PRIVATE KEY----- LINE1 LINE2 -----END EC PRIVATE KEY-----""", ), # credentials with human readable text ( "text1 text2 -----BEGIN EC PRIVATE KEY----- LINE1 LINE2 -----END EC PRIVATE KEY----- text1 text2", """text1 text2 -----BEGIN EC PRIVATE KEY----- LINE1 LINE2 -----END EC PRIVATE KEY----- text1 text2""", ), ], ) def test_replace_spaces_in_credentials(input_credentials, output_credentials): """ Given: Client cetifcates and private keys When: Authenticating the client Then: Check that the spaces in the credentials are replaced with new lines if they are in the correct format. """ from MailListenerV2 import replace_spaces_in_credentials import json assert json.dumps(replace_spaces_in_credentials(input_credentials)) == json.dumps(output_credentials) def test_fetch_incidents__last_uid_as_int(mocker): """ Given: - A mock client and last run with 'last_uid' as an integer - 8 When: - Fetching incidents Then: - Ensure that the "last_uid" received from the 'last_run' of previous cycles is converted to an integer. Also, verify that the 'last_uid' to be written in the 'last_run' for the next cycle is a string. """ from MailListenerV2 import fetch_incidents mocker.patch("MailListenerV2.Email.convert_to_incident", return_value={}) fetch_mail_mocker = mocker.patch("MailListenerV2.fetch_mails", return_value=([mock_email()], [mock_email()], 5)) next_run, _ = fetch_incidents( client=mocker.Mock(), last_run={"last_uid": 8}, first_fetch_time="2022-01-01 00:00:00", include_raw_body=False, with_headers=False, permitted_from_addresses="test@example.com", permitted_from_domains="example.com", delete_processed=False, limit=10, save_file=False, date_fetch=False, ) assert isinstance(fetch_mail_mocker.call_args[1]["uid_to_fetch_from"], int) assert isinstance(next_run["last_uid"], str) def test_fetch_incidents__last_uid_as_string(mocker): """ Given: - A mock client and last run with 'last_uid' as a string - "8" When: - Fetching incidents Then: - Ensure that the "last_uid" received from the 'last_run' of previous cycles is converted to an integer. Also, verify that the 'last_uid' to be written in the 'last_run' for the next cycle is a string. """ from MailListenerV2 import fetch_incidents mocker.patch("MailListenerV2.Email.convert_to_incident", return_value={}) fetch_mail_mocker = mocker.patch("MailListenerV2.fetch_mails", return_value=([mock_email()], [mock_email()], 5)) next_run, _ = fetch_incidents( client=mocker.Mock(), last_run={"last_uid": "8"}, first_fetch_time="2022-01-01 00:00:00", include_raw_body=False, with_headers=False, permitted_from_addresses="test@example.com", permitted_from_domains="example.com", delete_processed=False, limit=10, save_file=False, date_fetch=False, ) assert isinstance(fetch_mail_mocker.call_args[1]["uid_to_fetch_from"], int) assert isinstance(next_run["last_uid"], str) def test_fetch_incidents__last_uid_was_zero(mocker): """ Given: - A mock client and last run with 'last_uid' as a string - "0" When: - Fetching incidents Then: - Ensure that the next run is None, since setting it to "0" will cause an error in the next cycle. """ from MailListenerV2 import fetch_incidents mocker.patch("MailListenerV2.Email.convert_to_incident", return_value={}) mocker.patch("MailListenerV2.fetch_mails", return_value=([mock_email()], [mock_email()], 0)) next_run, _ = fetch_incidents( client=mocker.Mock(), last_run={"last_uid": "0"}, first_fetch_time="2022-01-01 00:00:00", include_raw_body=False, with_headers=False, permitted_from_addresses="test@example.com", permitted_from_domains="example.com", delete_processed=False, limit=10, save_file=False, date_fetch=False, ) assert next_run == {"last_date": "2022-01-01T00:00:00+00:00"} def test_fetch_incidents__email_without_date(mocker): """ Given: - A fetched email whose Date header could not be parsed (Email.date is None), together with a dated email, while fetching by date (date_fetch=True) with an existing last_date. When: - Fetching incidents. Then: - Ensure the fetch does not raise a TypeError from max(datetime, None), the undated email is still returned as an incident, and the date cursor advances to the newest available (dated) email. """ from MailListenerV2 import fetch_incidents mocker.patch("MailListenerV2.Email.convert_to_incident", return_value={}) dated_email = mock_email() dated_email.id = 6 undated_email = mock_email_without_date(mail_id=7) mocker.patch("MailListenerV2.fetch_mails", return_value=([dated_email, undated_email], [dated_email, undated_email], 7)) next_run, incidents = fetch_incidents( client=mocker.Mock(), last_run={"last_uid": "5", "last_date": "2020-01-01T00:00:00+00:00"}, first_fetch_time="2022-01-01 00:00:00", include_raw_body=False, with_headers=False, permitted_from_addresses="test@example.com", permitted_from_domains="example.com", delete_processed=False, limit=10, save_file=False, date_fetch=True, ) # Both emails are ingested (undated one is not dropped). assert len(incidents) == 2 # Cursor advanced to the dated email's date, ignoring the None date. assert next_run["last_date"] == "2020-10-01T00:00:00+00:00" assert next_run["last_uid"] == "7" def test_fetch_incidents__only_email_without_date(mocker): """ Given: - The only fetched email has no parseable Date header (Email.date is None), with date_fetch disabled and no prior last_date. In this first-fetch path time_to_fetch_from is initialized from first_fetch_time, while mail.date is None. When: - Fetching incidents. Then: - Ensure the fetch does not raise (the None mail.date is ignored in the max()), the undated email is still ingested, and the date cursor keeps the first_fetch_time value while the UID cursor advances. """ import demistomock as demisto from MailListenerV2 import fetch_incidents mocker.patch("MailListenerV2.Email.convert_to_incident", return_value={}) debug_mocker = mocker.patch.object(demisto, "debug") undated_email = mock_email_without_date(mail_id=7) mocker.patch("MailListenerV2.fetch_mails", return_value=([undated_email], [undated_email], 7)) next_run, incidents = fetch_incidents( client=mocker.Mock(), last_run={"last_uid": "5"}, first_fetch_time="2022-01-01 00:00:00", include_raw_body=False, with_headers=False, permitted_from_addresses="test@example.com", permitted_from_domains="example.com", delete_processed=False, limit=10, save_file=False, date_fetch=False, ) assert len(incidents) == 1 assert next_run["last_date"] == "2022-01-01T00:00:00+00:00" assert next_run["last_uid"] == "7" # Ensure we logged evidence that the email had no parseable Date header. assert any("has no parseable Date header" in str(call.args[0]) for call in debug_mocker.call_args_list if call.args) def test_fetch_mails__mail_id_is_greater(mocker): """ Given: - A mock client and last run with uid_to_fetch_from == 2 - The email UID returend from the client are [1, 2, 3] When: - Fetching incidents Then: - Ensure that next_uid_to_fetch_from is 3 since it is greater than the last run """ from MailListenerV2 import fetch_mails import demistomock as demisto from imapclient import IMAPClient mocker.patch("MailListenerV2.Email") mocker.patch.object(demisto, "debug") mocker.patch.object(IMAPClient, "search", return_value=[1, 2, 3]) mocker.patch.object(IMAPClient, "fetch") mocker.patch.object(IMAPClient, "_create_IMAP4") _, _, next_uid_to_fetch_from = fetch_mails(IMAPClient("http://example_url.com"), uid_to_fetch_from=2) assert next_uid_to_fetch_from == 3 def test_fetch_mails__last_run_is_greater(mocker): """ Given: - A mock client and last run with uid_to_fetch_from == 4 - The email UID returend from the client are [1, 2, 3] When: - Fetching incidents Then: - Ensure that the next_uid_to_fetch_from is 4 since it is greater than the greatest email UID """ from MailListenerV2 import fetch_mails import demistomock as demisto from imapclient import IMAPClient mocker.patch("MailListenerV2.Email") mocker.patch.object(demisto, "debug") mocker.patch.object(IMAPClient, "search", return_value=[1, 2, 3]) mocker.patch.object(IMAPClient, "fetch") mocker.patch.object(IMAPClient, "_create_IMAP4") _, _, next_uid_to_fetch_from = fetch_mails(IMAPClient("http://example_url.com"), uid_to_fetch_from=4) assert next_uid_to_fetch_from == 4 def test_fetch_mails__uid_is_str(mocker): """ Given: - The email UIDs returend from the client are strings ['1', '2', '3'] When: - Fetching incidents Then: - Ensure that the next_uid_to_fetch_from is 4 since it is greater than the greatest email UID """ from MailListenerV2 import fetch_mails import demistomock as demisto from imapclient import IMAPClient mocker.patch("MailListenerV2.Email") mocker.patch.object(demisto, "debug") mocker.patch.object(IMAPClient, "search", return_value=["1", "2", "3"]) mocker.patch.object(IMAPClient, "fetch") mocker.patch.object(IMAPClient, "_create_IMAP4") _, _, next_uid_to_fetch_from = fetch_mails(IMAPClient("http://example_url.com"), uid_to_fetch_from="4") assert next_uid_to_fetch_from == 4