Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def process_record(self, new, old=None):
new = super(Subscription, self).process_record(new, old)
try:
WebPusher(new)
except WebPushException as e:
raise http_error(HTTPBadRequest(),
errno=ERRORS.INVALID_PARAMETERS,
message='Invalid subscription: %s' % e)
return new
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
endpoint_domain = urlparse(subscription_spec['endpoint']).netloc
try:
result = webpush(
subscription_info=subscription_spec,
data=message_json,
timeout=10,
**vapid_params
)
pipe.send({
'subscription': subscription_id,
'endpoint_domain': endpoint_domain,
'result_status': str(result.status_code),
'result_message': result.text or None
})
except WebPushException as err:
pipe.send({
'subscription': subscription_id,
'endpoint_domain': endpoint_domain,
'result_status': 'WebPush Exception',
'result_message': str(err)
})
except ConnectTimeout as err:
pipe.send({
'subscription': subscription_id,
'endpoint_domain': endpoint_domain,
'result_status': 'Connection Timeout',
'result_message': str(err)
})
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
return
try:
response = webpush(
subscription_info=subscription_info,
data=message,
vapid_private_key=get_manager().get_wp_private_key(application_id),
vapid_claims=get_manager().get_wp_claims(application_id).copy(),
**kwargs
)
results = {"results": [{}]}
if not response.ok:
results["results"][0]["error"] = response.content
results["results"][0]["original_registration_id"] = response.content
else:
results["success"] = 1
return results
except WebPushException as e:
raise WebPushError(e.message)
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
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 Response("", 404)
return ""
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,
)
return None