github
/
apprise
mirror of https://github.com/caronc/apprise
1
0
Fork 0
Code Issues Releases Wiki Activity

refactored

pull/1225/head
Chris Caron 2024-11-30 21:12:21 -05:00
parent 5a3248147b
commit 2c6eb4a8b2
1 changed files with 53 additions and 97 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

150
apprise/plugins/office365.py
View File

@ -33,7 +33,11 @@
# Information on sending an email: # Information on sending an email:
# https://docs.microsoft.com/en-us/graph/api/user-sendmail\ # https://docs.microsoft.com/en-us/graph/api/user-sendmail\
# ?view=graph-rest-1.0&tabs=http # ?view=graph-rest-1.0&tabs=http
#
# Note: One must set up Application Permissions (not Delegated Permissions)
# - Scopes required: Mail.Send
# - For Large Attachments: Mail.ReadWrite
#
import requests import requests
from datetime import datetime from datetime import datetime
from datetime import timedelta from datetime import timedelta
@ -50,24 +54,6 @@ from ..utils import validate_regex
from ..locale import gettext_lazy as _ 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): class NotifyOffice365(NotifyBase):
""" """
A wrapper for Office 365 Notifications A wrapper for Office 365 Notifications
@ -117,11 +103,9 @@ class NotifyOffice365(NotifyBase):
# Define object templates # Define object templates
templates = ( templates = (
# Send as user # Send as user (only supported method)
'{schema}://{email}/{tenant}/{client_id}/{secret}', '{schema}://{source}/{tenant}/{client_id}/{secret}',
'{schema}://{email}/{tenant}/{client_id}/{secret}/{targets}', '{schema}://{source}/{tenant}/{client_id}/{secret}/{targets}',
# Send from 'me'
'{schema}://{tenant}/{client_id}/{secret}/{targets}',
) )
# Define our template tokens # Define our template tokens
@ -133,8 +117,8 @@ class NotifyOffice365(NotifyBase):
'private': True, 'private': True,
'regex': (r'^[a-z0-9-]+$', 'i'), 'regex': (r'^[a-z0-9-]+$', 'i'),
}, },
'email': { 'source': {
'name': _('Account Email'), 'name': _('Account Email or Object ID'),
'type': 'string', 'type': 'string',
'required': True, 'required': True,
}, },
@ -181,33 +165,15 @@ class NotifyOffice365(NotifyBase):
'oauth_secret': { 'oauth_secret': {
'alias_of': 'secret', 'alias_of': 'secret',
}, },
'mode': {
'name': _('Webhook Mode'),
'type': 'choice:string',
'values': OFFICE365_WEBHOOK_MODES,
'default': Office365WebhookMode.SELF,
},
}) })
def __init__(self, tenant, client_id, secret, email=None, def __init__(self, tenant, client_id, secret, source=None,
mode=None, targets=None, cc=None, bcc=None, **kwargs): targets=None, cc=None, bcc=None, **kwargs):
""" """
Initialize Office 365 Object Initialize Office 365 Object
""" """
super().__init__(**kwargs) 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 # Tenant identifier
self.tenant = validate_regex( self.tenant = validate_regex(
tenant, *self.template_tokens['tenant']['regex']) tenant, *self.template_tokens['tenant']['regex'])
@ -217,23 +183,8 @@ class NotifyOffice365(NotifyBase):
self.logger.warning(msg) self.logger.warning(msg)
raise TypeError(msg) raise TypeError(msg)
self.email = None # Store our email/ObjectID Source
if email is not None: self.source = source
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)
# Client Key (associated with generated OAuth2 Login) # Client Key (associated with generated OAuth2 Login)
self.client_id = validate_regex( self.client_id = validate_regex(
@ -280,8 +231,14 @@ class NotifyOffice365(NotifyBase):
.format(recipient)) .format(recipient))
else: else:
# If our target email list is empty we want to add ourselves to it result = is_email(self.source)
self.targets.append((False, self.email)) if not result:
self.logger.warning('No Target Office 365 Email Detected')
else:
# If our target email list is empty we want to add ourselves to
# it
self.targets.append((False, self.source))
# Validate recipients (cc:) and drop bad ones: # Validate recipients (cc:) and drop bad ones:
for recipient in parse_emails(cc): for recipient in parse_emails(cc):
@ -359,12 +316,9 @@ class NotifyOffice365(NotifyBase):
emails = list(self.targets) emails = list(self.targets)
# Define our URL to post to # Define our URL to post to
url = '{graph_url}/v1.0/me/sendMail'.format( url = '{graph_url}/v1.0/users/{userid}/sendMail'.format(
userid=self.source,
graph_url=self.graph_url, graph_url=self.graph_url,
) if not self.email \
else '{graph_url}/v1.0/users/{userid}/sendMail'.format(
userid=self.email,
graph_url=self.graph_url,
) )
attachments = [] attachments = []
@ -478,9 +432,7 @@ class NotifyOffice365(NotifyBase):
else '{}: '.format(self.names[e]), e) for e in bcc]))) else '{}: '.format(self.names[e]), e) for e in bcc])))
# Perform upstream fetch # Perform upstream fetch
postokay, response = self._fetch( postokay, response = self._fetch(url=url, payload=dumps(payload))
url=url, payload=dumps(payload),
content_type='application/json')
# Test if we were okay # Test if we were okay
if not postokay: if not postokay:
@ -513,12 +465,12 @@ class NotifyOffice365(NotifyBase):
# Prepare our payload # Prepare our payload
payload = { payload = {
'grant_type': 'client_credentials',
'client_id': self.client_id, 'client_id': self.client_id,
'client_secret': self.secret, 'client_secret': self.secret,
'scope': '{graph_url}/{scope}'.format( 'scope': '{graph_url}/{scope}'.format(
graph_url=self.graph_url, graph_url=self.graph_url,
scope=self.scope), scope=self.scope),
'grant_type': 'client_credentials',
} }
# Prepare our URL # Prepare our URL
@ -573,8 +525,7 @@ class NotifyOffice365(NotifyBase):
# We're authenticated # We're authenticated
return True if self.token else False return True if self.token else False
def _fetch(self, url, payload, def _fetch(self, url, payload, method='POST'):
content_type='application/x-www-form-urlencoded'):
""" """
Wrapper to request object Wrapper to request object
@ -583,7 +534,6 @@ class NotifyOffice365(NotifyBase):
# Prepare our headers: # Prepare our headers:
headers = { headers = {
'User-Agent': self.app_id, 'User-Agent': self.app_id,
'Content-Type': content_type,
} }
if self.token: if self.token:
@ -602,8 +552,9 @@ class NotifyOffice365(NotifyBase):
self.throttle() self.throttle()
# fetch function # fetch function
req = requests.post if method == 'POST' else requests.get
try: try:
r = requests.post( r = req(
url, url,
data=payload, data=payload,
headers=headers, headers=headers,
@ -625,6 +576,21 @@ class NotifyOffice365(NotifyBase):
', ' if status_str else '', ', ' if status_str else '',
r.status_code)) r.status_code))
# A Response could look like this if a Scope element was not
# found:
# {
# "error": {
# "code": "MissingClaimType",
# "message":"The token is missing the claim type \'oid\'.",
# "innerError": {
# "oAuthEventOperationId":" 7abe20-339f-4659-9381-38f52",
# "oAuthEventcV": "xsOSpAHSHVm3Tp4SNH5oIA.1.1",
# "errorUrl": "https://url",
# "requestId": "2328ea-ec9e-43a8-80f4-164c",
# "date":"2024-12-01T02:03:13"
# }}
# }
self.logger.debug( self.logger.debug(
'Response Details:\r\n{}'.format(r.content)) 'Response Details:\r\n{}'.format(r.content))
@ -659,7 +625,7 @@ class NotifyOffice365(NotifyBase):
here. here.
""" """
return ( return (
self.secure_protocol[0], self.email, self.tenant, self.client_id, self.secure_protocol[0], self.source, self.tenant, self.client_id,
self.secret, self.secret,
) )
@ -668,13 +634,8 @@ class NotifyOffice365(NotifyBase):
Returns the URL built dynamically based on specified arguments. Returns the URL built dynamically based on specified arguments.
""" """
# Define any URL parameters
params = {
'mode': self.mode,
}
# Extend our parameters # Extend our parameters
params.update(self.url_parameters(privacy=privacy, *args, **kwargs)) params = self.url_parameters(privacy=privacy, *args, **kwargs)
if self.cc: if self.cc:
# Handle our Carbon Copy Addresses # Handle our Carbon Copy Addresses
@ -690,13 +651,13 @@ class NotifyOffice365(NotifyBase):
'' if not self.names.get(e) '' if not self.names.get(e)
else '{}:'.format(self.names[e]), e) for e in self.bcc]) else '{}:'.format(self.names[e]), e) for e in self.bcc])
return '{schema}://{email}{tenant}/{client_id}/{secret}' \ return '{schema}://{source}/{tenant}/{client_id}/{secret}' \
'/{targets}/?{params}'.format( '/{targets}/?{params}'.format(
schema=self.secure_protocol[0], schema=self.secure_protocol[0],
tenant=self.pprint(self.tenant, privacy, safe=''), tenant=self.pprint(self.tenant, privacy, safe=''),
# email does not need to be escaped because it should # email does not need to be escaped because it should
# already be a valid host and username at this point # already be a valid host and username at this point
email=self.email + '/' if self.email else '', source=self.source,
client_id=self.pprint(self.client_id, privacy, safe=''), client_id=self.pprint(self.client_id, privacy, safe=''),
secret=self.pprint( secret=self.pprint(
self.secret, privacy, mode=PrivacyMode.Secret, self.secret, privacy, mode=PrivacyMode.Secret,
@ -743,20 +704,19 @@ class NotifyOffice365(NotifyBase):
if 'from' in results['qsd'] and \ if 'from' in results['qsd'] and \
len(results['qsd']['from']): len(results['qsd']['from']):
# Extract the sending account's information # Extract the sending account's information
results['email'] = \ results['source'] = \
NotifyOffice365.unquote(results['qsd']['from']) NotifyOffice365.unquote(results['qsd']['from'])
# If tenant is occupied, then the user defined makes up our email # If tenant is occupied, then the user defined makes up our source
elif results['user']: elif results['user']:
results['email'] = '{}@{}'.format( results['source'] = '{}@{}'.format(
NotifyOffice365.unquote(results['user']), NotifyOffice365.unquote(results['user']),
NotifyOffice365.unquote(results['host']), NotifyOffice365.unquote(results['host']),
) )
else: else:
# Hostname is no longer part of `from` and possibly instead # Object ID instead of email
# is the tenant id results['source'] = NotifyOffice365.unquote(results['host'])
entries.insert(0, NotifyOffice365.unquote(results['host']))
# Tenant # Tenant
if 'tenant' in results['qsd'] and len(results['qsd']['tenant']): if 'tenant' in results['qsd'] and len(results['qsd']['tenant']):
@ -825,8 +785,4 @@ class NotifyOffice365(NotifyBase):
if 'bcc' in results['qsd'] and len(results['qsd']['bcc']): if 'bcc' in results['qsd'] and len(results['qsd']['bcc']):
results['bcc'] = 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 return results