Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
vapid_data = {}
webpush_settings = getattr(settings, 'WEBPUSH_SETTINGS', {})
vapid_private_key = webpush_settings.get('VAPID_PRIVATE_KEY')
vapid_admin_email = webpush_settings.get('VAPID_ADMIN_EMAIL')
# Vapid keys are optional, and mandatory only for Chrome.
# If Vapid key is provided, include vapid key and claims
if vapid_private_key:
vapid_data = {
'vapid_private_key': vapid_private_key,
'vapid_claims': {"sub": "mailto:{}".format(vapid_admin_email)}
}
try:
req = webpush(subscription_info=subscription_data, data=payload, ttl=ttl, **vapid_data)
return req
except WebPushException as e:
# If the subscription is expired, delete it.
if e.response.status_code == 410:
subscription.delete()
else:
# Its other type of exception!
raise e
def notifyNewAgent():
dashboard_notifications = db.session.query(DashboardRegistration).all()
if dashboard_notifications is not None:
for registration in dashboard_notifications:
try:
webpush(
subscription_info={
"endpoint": registration.endpoint,
"keys": {
"p256dh": registration.authKey,
"auth": registration.authSecret
}
},
data="",
vapid_private_key="./private_key.pem",
vapid_claims={
"sub": "mailto:YourNameHere@example.org",
}
)
except WebPushException as ex:
print(ex)
return
def _send_webpush(subscription_info: dict, data: dict) -> bool:
"""
Send a webpush message asynchronously
"""
if not WEB_PUSH_ENABLED:
return False
json_data = json.dumps(data)
try:
webpush(
subscription_info=subscription_info,
data=json_data,
vapid_private_key=VAPID_PRIVATE_KEY,
vapid_claims=VAPID_CLAIMS,
)
return True
except WebPushException as error:
logger.error(error)
# Mozilla returns additional information in the body of the response.
if error.response and error.response.json():
logger.error(error.response.json())
return False
except TypeError as error:
logger.error(error)
raise error
async def send_web_push(subscription_information, message_body):
claims = dict(settings.VAPID_CLAIMS)
try:
return webpush(
subscription_info=subscription_information,
data=message_body,
vapid_private_key=settings.VAPID_PRIVATE_KEY,
vapid_claims=claims,
)
except WebPushException as ex:
logger.exception("Exception while trying to send push notification")
if ex.response and ex.response.json():
extra = ex.response.json()
logger.info(
"Remote service replied with a %s:%s, %s",
extra.code,
extra.errno,
extra.message,
)
def main():
""" Send data """
try:
args = get_config()
result = webpush(
args.sub_info,
data=args.data,
vapid_private_key=args.key,
vapid_claims=args.claims,
curl=args.curl,
content_encoding=args.encoding,
verbose=args.verbose)
print(result)
except Exception as ex:
print("ERROR: {}".format(ex))