From 5a3248147b00881da4222ed999d35f044603ffca Mon Sep 17 00:00:00 2001 From: Chris Caron Date: Sun, 27 Oct 2024 16:31:25 -0400 Subject: [PATCH] refactored code to parse URL correctly --- apprise/plugins/office365.py | 133 ++++++++++++++++++++++----------- test/test_plugin_office365.py | 135 +++++++++++++++++++++++++++++----- 2 files changed, 210 insertions(+), 58 deletions(-) diff --git a/apprise/plugins/office365.py b/apprise/plugins/office365.py index 65d191af0..b8b38ed6d 100644 --- a/apprise/plugins/office365.py +++ b/apprise/plugins/office365.py @@ -50,6 +50,24 @@ from ..locale import gettext_lazy as _ +class Office365WebhookMode: + """ + Office 365 Webhook Mode + """ + # Send message as ourselves using the /me/ endpoint + SELF = 'self' + + # Send message as ourselves using the /users/ endpoint + AS_USER = 'user' + + +# Define the modes in a list for validation purposes +OFFICE365_WEBHOOK_MODES = ( + Office365WebhookMode.SELF, + Office365WebhookMode.AS_USER, +) + + class NotifyOffice365(NotifyBase): """ A wrapper for Office 365 Notifications @@ -62,7 +80,7 @@ class NotifyOffice365(NotifyBase): service_url = 'https://office.com/' # The default protocol - secure_protocol = 'o365' + secure_protocol = ('azure', 'o365') # Allow 300 requests per minute. # 60/300 = 0.2 @@ -103,7 +121,6 @@ class NotifyOffice365(NotifyBase): '{schema}://{email}/{tenant}/{client_id}/{secret}', '{schema}://{email}/{tenant}/{client_id}/{secret}/{targets}', # Send from 'me' - '{schema}://{tenant}/{client_id}/{secret}', '{schema}://{tenant}/{client_id}/{secret}/{targets}', ) @@ -164,15 +181,33 @@ class NotifyOffice365(NotifyBase): 'oauth_secret': { 'alias_of': 'secret', }, + 'mode': { + 'name': _('Webhook Mode'), + 'type': 'choice:string', + 'values': OFFICE365_WEBHOOK_MODES, + 'default': Office365WebhookMode.SELF, + }, }) - def __init__(self, tenant, email, client_id, secret, - targets=None, cc=None, bcc=None, **kwargs): + def __init__(self, tenant, client_id, secret, email=None, + mode=None, targets=None, cc=None, bcc=None, **kwargs): """ Initialize Office 365 Object """ super().__init__(**kwargs) + # Prepare our Mode + self.mode = self.template_args['mode']['default'] \ + if not mode else next( + (f for f in OFFICE365_WEBHOOK_MODES + if f.startswith( + mode.lower())), None) + if mode and not self.mode: + msg = \ + 'The specified Webhook mode ({}) was not found '.format(mode) + self.logger.warning(msg) + raise TypeError(msg) + # Tenant identifier self.tenant = validate_regex( tenant, *self.template_tokens['tenant']['regex']) @@ -182,16 +217,24 @@ def __init__(self, tenant, email, client_id, secret, self.logger.warning(msg) raise TypeError(msg) - result = is_email(email) - if not result: - msg = 'An invalid Office 365 Email Account ID' \ - '({}) was specified.'.format(email) + self.email = None + if email is not None: + result = is_email(email) + if not result: + msg = 'An invalid Office 365 Email Account ID' \ + '({}) was specified.'.format(email) + self.logger.warning(msg) + raise TypeError(msg) + + # Otherwise store our the email address + self.email = result['full_email'] + + elif self.mode != Office365WebhookMode.SELF: + msg = 'An expected Office 365 Email was not specified ' \ + '(mode={})'.format(self.mode) self.logger.warning(msg) raise TypeError(msg) - # Otherwise store our the email address - self.email = result['full_email'] - # Client Key (associated with generated OAuth2 Login) self.client_id = validate_regex( client_id, *self.template_tokens['client_id']['regex']) @@ -318,7 +361,7 @@ def send(self, body, title='', notify_type=NotifyType.INFO, attach=None, # Define our URL to post to url = '{graph_url}/v1.0/me/sendMail'.format( graph_url=self.graph_url, - ) if not self.self.email \ + ) if not self.email \ else '{graph_url}/v1.0/users/{userid}/sendMail'.format( userid=self.email, graph_url=self.graph_url, @@ -616,7 +659,7 @@ def url_identifier(self): here. """ return ( - self.secure_protocol, self.email, self.tenant, self.client_id, + self.secure_protocol[0], self.email, self.tenant, self.client_id, self.secret, ) @@ -625,8 +668,13 @@ def url(self, privacy=False, *args, **kwargs): Returns the URL built dynamically based on specified arguments. """ - # Our URL parameters - params = self.url_parameters(privacy=privacy, *args, **kwargs) + # Define any URL parameters + params = { + 'mode': self.mode, + } + + # Extend our parameters + params.update(self.url_parameters(privacy=privacy, *args, **kwargs)) if self.cc: # Handle our Carbon Copy Addresses @@ -642,13 +690,13 @@ def url(self, privacy=False, *args, **kwargs): '' if not self.names.get(e) else '{}:'.format(self.names[e]), e) for e in self.bcc]) - return '{schema}://{tenant}:{email}/{client_id}/{secret}' \ + return '{schema}://{email}{tenant}/{client_id}/{secret}' \ '/{targets}/?{params}'.format( - schema=self.secure_protocol, + schema=self.secure_protocol[0], tenant=self.pprint(self.tenant, privacy, safe=''), # email does not need to be escaped because it should # already be a valid host and username at this point - email=self.email, + email=self.email + '/' if self.email else '', client_id=self.pprint(self.client_id, privacy, safe=''), secret=self.pprint( self.secret, privacy, mode=PrivacyMode.Secret, @@ -656,7 +704,7 @@ def url(self, privacy=False, *args, **kwargs): targets='/'.join( [NotifyOffice365.quote('{}{}'.format( '' if not e[0] else '{}:'.format(e[0]), e[1]), - safe='') for e in self.targets]), + safe='@') for e in self.targets]), params=NotifyOffice365.urlencode(params)) def __len__(self): @@ -687,6 +735,7 @@ def parse_url(url): # Initialize our tenant results['tenant'] = None + # Initialize our email results['email'] = None @@ -697,28 +746,36 @@ def parse_url(url): results['email'] = \ NotifyOffice365.unquote(results['qsd']['from']) + # If tenant is occupied, then the user defined makes up our email + elif results['user']: + results['email'] = '{}@{}'.format( + NotifyOffice365.unquote(results['user']), + NotifyOffice365.unquote(results['host']), + ) + + else: # Hostname is no longer part of `from` and possibly instead # is the tenant id entries.insert(0, NotifyOffice365.unquote(results['host'])) # Tenant - if 'tenant' in results['qsd'] and \ - len(results['qsd']['tenant']): + if 'tenant' in results['qsd'] and len(results['qsd']['tenant']): # Extract the Tenant from the argument results['tenant'] = \ NotifyOffice365.unquote(results['qsd']['tenant']) - # If tenant is occupied, then the user defined makes - # up our email - if not results['email'] and results['user']: - results['email'] = '{}@{}'.format( - NotifyOffice365.unquote(results['user']), - NotifyOffice365.unquote(results['host']), - ) + elif entries: + results['tenant'] = NotifyOffice365.unquote(entries.pop(0)) + + # OAuth2 ID + if 'oauth_id' in results['qsd'] and len(results['qsd']['oauth_id']): + # Extract the API Key from an argument + results['client_id'] = \ + NotifyOffice365.unquote(results['qsd']['oauth_id']) - elif not results['user']: - # Only tenant id specified (emails are sent 'from me') - results['tenant'] = NotifyOffice365.unquote(results['host']) + elif entries: + # Get our client_id is the first entry on the path + results['client_id'] = NotifyOffice365.unquote(entries.pop(0)) # # Prepare our target listing @@ -740,16 +797,6 @@ def parse_url(url): # We're done break - # OAuth2 ID - if 'oauth_id' in results['qsd'] and len(results['qsd']['oauth_id']): - # Extract the API Key from an argument - results['client_id'] = \ - NotifyOffice365.unquote(results['qsd']['oauth_id']) - - elif entries: - # Get our client_id is the first entry on the path - results['client_id'] = NotifyOffice365.unquote(entries.pop(0)) - # OAuth2 Secret if 'oauth_secret' in results['qsd'] and \ len(results['qsd']['oauth_secret']): @@ -778,4 +825,8 @@ def parse_url(url): if 'bcc' in results['qsd'] and len(results['qsd']['bcc']): results['bcc'] = results['qsd']['bcc'] + # Handle Mode + if 'mode' in results['qsd'] and len(results['qsd']['mode']): + results['mode'] = results['qsd']['mode'] + return results diff --git a/test/test_plugin_office365.py b/test/test_plugin_office365.py index cca9e1e99..c22d38384 100644 --- a/test/test_plugin_office365.py +++ b/test/test_plugin_office365.py @@ -34,6 +34,8 @@ from datetime import datetime from json import dumps from apprise import Apprise +from apprise import NotifyType +from apprise import AppriseAttachment from apprise.plugins.office365 import NotifyOffice365 from helpers import AppriseURLTester @@ -57,7 +59,7 @@ # invalid url 'instance': TypeError, }), - ('o365://{tenant}:{aid}/{cid}/{secret}/{targets}'.format( + ('o365://{aid}/{tenant}/{cid}/{secret}/{targets}'.format( # invalid tenant tenant=',', cid='ab-cd-ef-gh', @@ -65,10 +67,10 @@ secret='abcd/123/3343/@jack/test', targets='/'.join(['email1@test.ca'])), { - # We're valid and good to go + # Expected failure 'instance': TypeError, }), - ('o365://{tenant}:{aid}/{cid}/{secret}/{targets}'.format( + ('o365://{aid}/{tenant}/{cid}/{secret}/{targets}'.format( tenant='tenant', # invalid client id cid='ab.', @@ -76,16 +78,53 @@ secret='abcd/123/3343/@jack/test', targets='/'.join(['email1@test.ca'])), { - # We're valid and good to go + # Expected failure 'instance': TypeError, }), - ('o365://{tenant}:{aid}/{cid}/{secret}/{targets}'.format( + ('o365://{aid}/{tenant}/{cid}/{secret}/{targets}?mode=invalid'.format( + # Invalid mode tenant='tenant', cid='ab-cd-ef-gh', aid='user@example.com', secret='abcd/123/3343/@jack/test', targets='/'.join(['email1@test.ca'])), { + # Expected failure + 'instance': TypeError, + }), + ('o365://{tenant}/{cid}/{secret}/{targets}?mode=user'.format( + # Invalid mode when no email specified + tenant='tenant', + cid='ab-cd-ef-gh', + secret='abcd/123/3343/@jack/test', + targets='/'.join(['email1@test.ca'])), { + + # Expected failure + 'instance': TypeError, + }), + ('o365://{tenant}/{cid}/{secret}/{targets}?mode=self'.format( + # email not required if mode is set to self + tenant='tenant', + cid='ab-cd-ef-gh', + secret='abcd/123/3343/@jack/test', + targets='/'.join(['email1@test.ca'])), { + + # We're valid and good to go + 'instance': NotifyOffice365, + + # Test what happens if a batch send fails to return a messageCount + 'requests_response_text': { + 'expires_in': 2000, + 'access_token': 'abcd1234', + }, + }), + ('o365://{aid}/{tenant}/{cid}/{secret}/{targets}'.format( + tenant='tenant', + cid='ab-cd-ef-gh', + aid='user@example.edu', + secret='abcd/123/3343/@jack/test', + targets='/'.join(['email1@test.ca'])), { + # We're valid and good to go 'instance': NotifyOffice365, @@ -96,14 +135,14 @@ }, # Our expected url(privacy=True) startswith() response: - 'privacy_url': 'o365://t...t:user@example.com/a...h/' - '****/email1%40test.ca/'}), + 'privacy_url': 'azure://user@example.edu/t...t/a...h/' + '****/email1@test.ca/'}), # test our arguments ('o365://_/?oauth_id={cid}&oauth_secret={secret}&tenant={tenant}' '&to={targets}&from={aid}'.format( tenant='tenant', cid='ab-cd-ef-gh', - aid='user@example.com', + aid='user@example.ca', secret='abcd/123/3343/@jack/test', targets='email1@test.ca'), { @@ -117,10 +156,10 @@ }, # Our expected url(privacy=True) startswith() response: - 'privacy_url': 'o365://t...t:user@example.com/a...h/' - '****/email1%40test.ca/'}), + 'privacy_url': 'azure://user@example.ca/t...t/a...h/' + '****/email1@test.ca/'}), # Test invalid JSON (no tenant defaults to email domain) - ('o365://{tenant}:{aid}/{cid}/{secret}/{targets}'.format( + ('o365://{aid}/{tenant}/{cid}/{secret}/{targets}'.format( tenant='tenant', cid='ab-cd-ef-gh', aid='user@example.com', @@ -135,7 +174,7 @@ 'notify_response': False, }), # No Targets specified - ('o365://{tenant}:{aid}/{cid}/{secret}'.format( + ('o365://{aid}/{tenant}/{cid}/{secret}'.format( tenant='tenant', cid='ab-cd-ef-gh', aid='user@example.com', @@ -150,7 +189,7 @@ 'access_token': 'abcd1234', }, }), - ('o365://{tenant}:{aid}/{cid}/{secret}/{targets}'.format( + ('o365://{aid}/{tenant}/{cid}/{secret}/{targets}'.format( tenant='tenant', cid='zz-zz-zz-zz', aid='user@example.com', @@ -212,7 +251,7 @@ def test_plugin_office365_general(mock_post): # Instantiate our object obj = Apprise.instantiate( - 'o365://{tenant}:{email}/{tenant}/{secret}/{targets}'.format( + 'o365://{email}/{tenant}/{secret}/{targets}'.format( tenant=tenant, email=email, secret=secret, @@ -228,10 +267,11 @@ def test_plugin_office365_general(mock_post): # Instantiate our object obj = Apprise.instantiate( - 'o365://{tenant}:{email}/{tenant}/{secret}/{targets}' + 'o365://{email}/{tenant}/{client_id}/{secret}/{targets}' '?bcc={bcc}&cc={cc}'.format( tenant=tenant, email=email, + client_id=client_id, secret=secret, targets=targets, # Test the cc and bcc list (use good and bad email) @@ -260,7 +300,7 @@ def test_plugin_office365_general(mock_post): with pytest.raises(TypeError): # Invalid email NotifyOffice365( - email=None, + email='invalid', client_id=client_id, tenant=tenant, secret=secret, @@ -336,7 +376,7 @@ def test_plugin_office365_authentication(mock_post): # Instantiate our object obj = Apprise.instantiate( - 'o365://{tenant}:{email}/{client_id}/{secret}/{targets}'.format( + 'azure://{email}/{tenant}/{client_id}/{secret}/{targets}'.format( client_id=client_id, tenant=tenant, email=email, @@ -394,3 +434,64 @@ def test_plugin_office365_authentication(mock_post): del invalid_auth_entries['expires_in'] response.content = dumps(invalid_auth_entries) assert obj.authenticate() is False + + +@mock.patch('requests.post') +def test_plugin_office365_attachments(mock_post): + """ + NotifyOffice365() Attachments + + """ + + # Initialize some generic (but valid) tokens + email = 'user@example.net' + tenant = 'ff-gg-hh-ii-jj' + client_id = 'aa-bb-cc-dd-ee' + secret = 'abcd/1234/abcd@ajd@/test' + targets = 'target@example.com' + + # Prepare Mock return object + authentication = { + "token_type": "Bearer", + "expires_in": 6000, + "access_token": "abcd1234" + } + okay_response = mock.Mock() + okay_response.content = dumps(authentication) + okay_response.status_code = requests.codes.ok + mock_post.return_value = okay_response + + # Instantiate our object + obj = Apprise.instantiate( + 'azure://{email}/{tenant}/{client_id}{secret}/{targets}'.format( + client_id=client_id, + tenant=tenant, + email=email, + secret=secret, + targets=targets)) + + assert isinstance(obj, NotifyOffice365) + + # Test Valid Attachment + path = os.path.join(TEST_VAR_DIR, 'apprise-test.gif') + attach = AppriseAttachment(path) + assert obj.notify( + body='body', title='title', notify_type=NotifyType.INFO, + attach=attach) is True + + # Test invalid attachment + path = os.path.join(TEST_VAR_DIR, '/invalid/path/to/an/invalid/file.jpg') + assert obj.notify( + body='body', title='title', notify_type=NotifyType.INFO, + attach=path) is False + + with mock.patch('base64.b64encode', side_effect=OSError()): + # We can't send the message if we fail to parse the data + assert obj.notify( + body='body', title='title', notify_type=NotifyType.INFO, + attach=attach) is False + + assert obj.notify( + body='body', title='title', notify_type=NotifyType.INFO, + attach=attach) is True + assert mock_post.call_count == 3