Home-Assistant
/
core
mirror of https://github.com/home-assistant/core.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Fix html5 Firefox Notifications (#82556) 

Co-authored-by: Paulus Schoutsen <paulus@home-assistant.io>
fixes undefined
pull/83170/head
Alex Yao 2022-12-02 19:51:16 -06:00 committed by GitHub
parent ee0fbae2ca
commit 652fedf4d1
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 59 additions and 55 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

25
homeassistant/components/html5/notify.py
View File

@ -99,6 +99,7 @@ SCHEMA_WS_APPKEY = websocket_api.BASE_COMMAND_MESSAGE_SCHEMA.extend(
# The number of days after the moment a notification is sent that a JWT
# is valid.
JWT_VALID_DAYS = 7
VAPID_CLAIM_VALID_HOURS = 12
KEYS_SCHEMA = vol.All(
dict,
@ -514,7 +515,10 @@ class HTML5NotificationService(BaseNotificationService):
webpusher = WebPusher(info[ATTR_SUBSCRIPTION])
if self._vapid_prv and self._vapid_email:
vapid_headers = create_vapid_headers(
self._vapid_email, info[ATTR_SUBSCRIPTION], self._vapid_prv
self._vapid_email,
info[ATTR_SUBSCRIPTION],
self._vapid_prv,
timestamp,
)
vapid_headers.update({"urgency": priority, "priority": priority})
response = webpusher.send(
@ -540,6 +544,12 @@ class HTML5NotificationService(BaseNotificationService):
_LOGGER.error("Error saving registration")
else:
_LOGGER.info("Configuration saved")
elif response.status_code > 399:
_LOGGER.error(
"There was an issue sending the notification %s: %s",
response.status,
response.text,
)
def add_jwt(timestamp, target, tag, jwt_secret):
@ -556,14 +566,23 @@ def add_jwt(timestamp, target, tag, jwt_secret):
return jwt.encode(jwt_claims, jwt_secret)
def create_vapid_headers(vapid_email, subscription_info, vapid_private_key):
def create_vapid_headers(vapid_email, subscription_info, vapid_private_key, timestamp):
"""Create encrypted headers to send to WebPusher."""
if vapid_email and vapid_private_key and ATTR_ENDPOINT in subscription_info:
if (
vapid_email
and vapid_private_key
and ATTR_ENDPOINT in subscription_info
and timestamp
):
vapid_exp = datetime.fromtimestamp(timestamp) + timedelta(
hours=VAPID_CLAIM_VALID_HOURS
)
url = urlparse(subscription_info.get(ATTR_ENDPOINT))
vapid_claims = {
"sub": f"mailto:{vapid_email}",
"aud": f"{url.scheme}://{url.netloc}",
"exp": int(vapid_exp.timestamp()),
}
vapid = Vapid.from_string(private_key=vapid_private_key)
return vapid.sign(vapid_claims)

89
tests/components/html5/test_notify.py
View File

@ -93,6 +93,7 @@ class TestHtml5Notify:
def test_dismissing_message(self, mock_wp):
"""Test dismissing message."""
hass = MagicMock()
mock_wp().send().status_code = 201
data = {"device": SUBSCRIPTION_1}
@ -104,15 +105,13 @@ class TestHtml5Notify:
service.dismiss(target=["device", "non_existing"], data={"tag": "test"})
assert len(mock_wp.mock_calls) == 3
assert len(mock_wp.mock_calls) == 4
# WebPusher constructor
assert mock_wp.mock_calls[0][1][0] == SUBSCRIPTION_1["subscription"]
# Third mock_call checks the status_code of the response.
assert mock_wp.mock_calls[2][0] == "().send().status_code.__eq__"
assert mock_wp.mock_calls[2][1][0] == SUBSCRIPTION_1["subscription"]
# Call to send
payload = json.loads(mock_wp.mock_calls[1][1][0])
payload = json.loads(mock_wp.mock_calls[3][1][0])
assert payload["dismiss"] is True
assert payload["tag"] == "test"
@ -121,6 +120,7 @@ class TestHtml5Notify:
def test_sending_message(self, mock_wp):
"""Test sending message."""
hass = MagicMock()
mock_wp().send().status_code = 201
data = {"device": SUBSCRIPTION_1}
@ -134,15 +134,13 @@ class TestHtml5Notify:
"Hello", target=["device", "non_existing"], data={"icon": "beer.png"}
)
assert len(mock_wp.mock_calls) == 3
assert len(mock_wp.mock_calls) == 4
# WebPusher constructor
assert mock_wp.mock_calls[0][1][0] == SUBSCRIPTION_1["subscription"]
# Third mock_call checks the status_code of the response.
assert mock_wp.mock_calls[2][0] == "().send().status_code.__eq__"
assert mock_wp.mock_calls[2][1][0] == SUBSCRIPTION_1["subscription"]
# Call to send
payload = json.loads(mock_wp.mock_calls[1][1][0])
payload = json.loads(mock_wp.mock_calls[3][1][0])
assert payload["body"] == "Hello"
assert payload["icon"] == "beer.png"
@ -151,6 +149,7 @@ class TestHtml5Notify:
def test_gcm_key_include(self, mock_wp):
"""Test if the gcm_key is only included for GCM endpoints."""
hass = MagicMock()
mock_wp().send().status_code = 201
data = {"chrome": SUBSCRIPTION_1, "firefox": SUBSCRIPTION_2}
@ -167,21 +166,18 @@ class TestHtml5Notify:
assert len(mock_wp.mock_calls) == 6
# WebPusher constructor
assert mock_wp.mock_calls[0][1][0] == SUBSCRIPTION_1["subscription"]
assert mock_wp.mock_calls[3][1][0] == SUBSCRIPTION_2["subscription"]
# Third mock_call checks the status_code of the response.
assert mock_wp.mock_calls[2][0] == "().send().status_code.__eq__"
assert mock_wp.mock_calls[5][0] == "().send().status_code.__eq__"
assert mock_wp.mock_calls[2][1][0] == SUBSCRIPTION_1["subscription"]
assert mock_wp.mock_calls[4][1][0] == SUBSCRIPTION_2["subscription"]
# Get the keys passed to the WebPusher's send method
assert mock_wp.mock_calls[1][2]["gcm_key"] is not None
assert mock_wp.mock_calls[4][2]["gcm_key"] is None
assert mock_wp.mock_calls[3][2]["gcm_key"] is not None
assert mock_wp.mock_calls[5][2]["gcm_key"] is None
@patch("homeassistant.components.html5.notify.WebPusher")
def test_fcm_key_include(self, mock_wp):
"""Test if the FCM header is included."""
hass = MagicMock()
mock_wp().send().status_code = 201
data = {"chrome": SUBSCRIPTION_5}
@ -193,20 +189,18 @@ class TestHtml5Notify:
service.send_message("Hello", target=["chrome"])
assert len(mock_wp.mock_calls) == 3
assert len(mock_wp.mock_calls) == 4
# WebPusher constructor
assert mock_wp.mock_calls[0][1][0] == SUBSCRIPTION_5["subscription"]
# Third mock_call checks the status_code of the response.
assert mock_wp.mock_calls[2][0] == "().send().status_code.__eq__"
assert mock_wp.mock_calls[2][1][0] == SUBSCRIPTION_5["subscription"]
# Get the keys passed to the WebPusher's send method
assert mock_wp.mock_calls[1][2]["headers"]["Authorization"] is not None
assert mock_wp.mock_calls[3][2]["headers"]["Authorization"] is not None
@patch("homeassistant.components.html5.notify.WebPusher")
def test_fcm_send_with_unknown_priority(self, mock_wp):
"""Test if the gcm_key is only included for GCM endpoints."""
hass = MagicMock()
mock_wp().send().status_code = 201
data = {"chrome": SUBSCRIPTION_5}
@ -218,20 +212,18 @@ class TestHtml5Notify:
service.send_message("Hello", target=["chrome"], priority="undefined")
assert len(mock_wp.mock_calls) == 3
assert len(mock_wp.mock_calls) == 4
# WebPusher constructor
assert mock_wp.mock_calls[0][1][0] == SUBSCRIPTION_5["subscription"]
# Third mock_call checks the status_code of the response.
assert mock_wp.mock_calls[2][0] == "().send().status_code.__eq__"
assert mock_wp.mock_calls[2][1][0] == SUBSCRIPTION_5["subscription"]
# Get the keys passed to the WebPusher's send method
assert mock_wp.mock_calls[1][2]["headers"]["priority"] == "normal"
assert mock_wp.mock_calls[3][2]["headers"]["priority"] == "normal"
@patch("homeassistant.components.html5.notify.WebPusher")
def test_fcm_no_targets(self, mock_wp):
"""Test if the gcm_key is only included for GCM endpoints."""
hass = MagicMock()
mock_wp().send().status_code = 201
data = {"chrome": SUBSCRIPTION_5}
@ -243,20 +235,18 @@ class TestHtml5Notify:
service.send_message("Hello")
assert len(mock_wp.mock_calls) == 3
assert len(mock_wp.mock_calls) == 4
# WebPusher constructor
assert mock_wp.mock_calls[0][1][0] == SUBSCRIPTION_5["subscription"]
# Third mock_call checks the status_code of the response.
assert mock_wp.mock_calls[2][0] == "().send().status_code.__eq__"
assert mock_wp.mock_calls[2][1][0] == SUBSCRIPTION_5["subscription"]
# Get the keys passed to the WebPusher's send method
assert mock_wp.mock_calls[1][2]["headers"]["priority"] == "normal"
assert mock_wp.mock_calls[3][2]["headers"]["priority"] == "normal"
@patch("homeassistant.components.html5.notify.WebPusher")
def test_fcm_additional_data(self, mock_wp):
"""Test if the gcm_key is only included for GCM endpoints."""
hass = MagicMock()
mock_wp().send().status_code = 201
data = {"chrome": SUBSCRIPTION_5}
@ -268,21 +258,18 @@ class TestHtml5Notify:
service.send_message("Hello", data={"mykey": "myvalue"})
assert len(mock_wp.mock_calls) == 3
assert len(mock_wp.mock_calls) == 4
# WebPusher constructor
assert mock_wp.mock_calls[0][1][0] == SUBSCRIPTION_5["subscription"]
# Third mock_call checks the status_code of the response.
assert mock_wp.mock_calls[2][0] == "().send().status_code.__eq__"
assert mock_wp.mock_calls[2][1][0] == SUBSCRIPTION_5["subscription"]
# Get the keys passed to the WebPusher's send method
assert mock_wp.mock_calls[1][2]["headers"]["priority"] == "normal"
assert mock_wp.mock_calls[3][2]["headers"]["priority"] == "normal"
def test_create_vapid_withoutvapid():
"""Test creating empty vapid."""
resp = html5.create_vapid_headers(
vapid_email=None, vapid_private_key=None, subscription_info=None
vapid_email=None, vapid_private_key=None, subscription_info=None, timestamp=None
)
assert resp is None
@ -478,6 +465,7 @@ async def test_callback_view_with_jwt(hass, hass_client):
client = await mock_client(hass, hass_client, registrations)
with patch("homeassistant.components.html5.notify.WebPusher") as mock_wp:
mock_wp().send().status_code = 201
await hass.services.async_call(
"notify",
"notify",
@ -485,15 +473,13 @@ async def test_callback_view_with_jwt(hass, hass_client):
blocking=True,
)
assert len(mock_wp.mock_calls) == 3
assert len(mock_wp.mock_calls) == 4
# WebPusher constructor
assert mock_wp.mock_calls[0][1][0] == SUBSCRIPTION_1["subscription"]
# Third mock_call checks the status_code of the response.
assert mock_wp.mock_calls[2][0] == "().send().status_code.__eq__"
assert mock_wp.mock_calls[2][1][0] == SUBSCRIPTION_1["subscription"]
# Call to send
push_payload = json.loads(mock_wp.mock_calls[1][1][0])
push_payload = json.loads(mock_wp.mock_calls[3][1][0])
assert push_payload["body"] == "Hello"
assert push_payload["icon"] == "beer.png"
@ -514,6 +500,7 @@ async def test_send_fcm_without_targets(hass, hass_client):
registrations = {"device": SUBSCRIPTION_5}
await mock_client(hass, hass_client, registrations)
with patch("homeassistant.components.html5.notify.WebPusher") as mock_wp:
mock_wp().send().status_code = 201
await hass.services.async_call(
"notify",
"notify",
@ -521,12 +508,10 @@ async def test_send_fcm_without_targets(hass, hass_client):
blocking=True,
)
assert len(mock_wp.mock_calls) == 3
assert len(mock_wp.mock_calls) == 4
# WebPusher constructor
assert mock_wp.mock_calls[0][1][0] == SUBSCRIPTION_5["subscription"]
# Third mock_call checks the status_code of the response.
assert mock_wp.mock_calls[2][0] == "().send().status_code.__eq__"
assert mock_wp.mock_calls[2][1][0] == SUBSCRIPTION_5["subscription"]
async def test_send_fcm_expired(hass, hass_client):