"""Webhook handlers for mobile_app.""" from functools import partial import logging from typing import Dict from aiohttp.web import HTTPBadRequest, Response, Request import voluptuous as vol from homeassistant.components.device_tracker import (ATTR_ATTRIBUTES, ATTR_DEV_ID, DOMAIN as DT_DOMAIN, SERVICE_SEE as DT_SEE) from homeassistant.components.webhook import async_register as webhook_register from homeassistant.const import (ATTR_DOMAIN, ATTR_SERVICE, ATTR_SERVICE_DATA, CONF_WEBHOOK_ID, HTTP_BAD_REQUEST) from homeassistant.core import EventOrigin from homeassistant.exceptions import (HomeAssistantError, ServiceNotFound, TemplateError) from homeassistant.helpers.template import attach from homeassistant.helpers.discovery import load_platform from homeassistant.helpers.storage import Store from homeassistant.helpers.typing import HomeAssistantType from .const import (ATTR_ALTITUDE, ATTR_APP_COMPONENT, ATTR_BATTERY, ATTR_COURSE, ATTR_DEVICE_ID, ATTR_DEVICE_NAME, ATTR_EVENT_DATA, ATTR_EVENT_TYPE, ATTR_GPS, ATTR_GPS_ACCURACY, ATTR_LOCATION_NAME, ATTR_SPEED, ATTR_SUPPORTS_ENCRYPTION, ATTR_TEMPLATE, ATTR_TEMPLATE_VARIABLES, ATTR_VERTICAL_ACCURACY, ATTR_WEBHOOK_DATA, ATTR_WEBHOOK_ENCRYPTED, ATTR_WEBHOOK_ENCRYPTED_DATA, ATTR_WEBHOOK_TYPE, CONF_SECRET, DATA_DELETED_IDS, DATA_REGISTRATIONS, DOMAIN, ERR_ENCRYPTION_REQUIRED, WEBHOOK_PAYLOAD_SCHEMA, WEBHOOK_SCHEMAS, WEBHOOK_TYPE_CALL_SERVICE, WEBHOOK_TYPE_FIRE_EVENT, WEBHOOK_TYPE_RENDER_TEMPLATE, WEBHOOK_TYPE_UPDATE_LOCATION, WEBHOOK_TYPE_UPDATE_REGISTRATION) from .helpers import (_decrypt_payload, empty_okay_response, error_response, registration_context, safe_registration, savable_state, webhook_response) _LOGGER = logging.getLogger(__name__) def register_deleted_webhooks(hass: HomeAssistantType, store: Store): """Register previously deleted webhook IDs so we can return 410.""" for deleted_id in hass.data[DOMAIN][DATA_DELETED_IDS]: try: webhook_register(hass, DOMAIN, "Deleted Webhook", deleted_id, partial(handle_webhook, store)) except ValueError: pass def setup_registration(hass: HomeAssistantType, store: Store, registration: Dict) -> None: """Register the webhook for a registration and loads the app component.""" registration_name = 'Mobile App: {}'.format(registration[ATTR_DEVICE_NAME]) webhook_id = registration[CONF_WEBHOOK_ID] webhook_register(hass, DOMAIN, registration_name, webhook_id, partial(handle_webhook, store)) if ATTR_APP_COMPONENT in registration: load_platform(hass, registration[ATTR_APP_COMPONENT], DOMAIN, {}, {DOMAIN: {}}) async def handle_webhook(store: Store, hass: HomeAssistantType, webhook_id: str, request: Request) -> Response: """Handle webhook callback.""" if webhook_id in hass.data[DOMAIN][DATA_DELETED_IDS]: return Response(status=410) headers = {} registration = hass.data[DOMAIN][DATA_REGISTRATIONS][webhook_id] try: req_data = await request.json() except ValueError: _LOGGER.warning('Received invalid JSON from mobile_app') return empty_okay_response(status=HTTP_BAD_REQUEST) if (ATTR_WEBHOOK_ENCRYPTED not in req_data and registration[ATTR_SUPPORTS_ENCRYPTION]): _LOGGER.warning("Refusing to accept unencrypted webhook from %s", registration[ATTR_DEVICE_NAME]) return error_response(ERR_ENCRYPTION_REQUIRED, "Encryption required") try: req_data = WEBHOOK_PAYLOAD_SCHEMA(req_data) except vol.Invalid as ex: err = vol.humanize.humanize_error(req_data, ex) _LOGGER.error('Received invalid webhook payload: %s', err) return empty_okay_response() webhook_type = req_data[ATTR_WEBHOOK_TYPE] webhook_payload = req_data.get(ATTR_WEBHOOK_DATA, {}) if req_data[ATTR_WEBHOOK_ENCRYPTED]: enc_data = req_data[ATTR_WEBHOOK_ENCRYPTED_DATA] webhook_payload = _decrypt_payload(registration[CONF_SECRET], enc_data) try: data = WEBHOOK_SCHEMAS[webhook_type](webhook_payload) except vol.Invalid as ex: err = vol.humanize.humanize_error(webhook_payload, ex) _LOGGER.error('Received invalid webhook payload: %s', err) return empty_okay_response(headers=headers) context = registration_context(registration) if webhook_type == WEBHOOK_TYPE_CALL_SERVICE: try: await hass.services.async_call(data[ATTR_DOMAIN], data[ATTR_SERVICE], data[ATTR_SERVICE_DATA], blocking=True, context=context) # noqa: E722 pylint: disable=broad-except except (vol.Invalid, ServiceNotFound, Exception) as ex: _LOGGER.error("Error when calling service during mobile_app " "webhook (device name: %s): %s", registration[ATTR_DEVICE_NAME], ex) raise HTTPBadRequest() return empty_okay_response(headers=headers) if webhook_type == WEBHOOK_TYPE_FIRE_EVENT: event_type = data[ATTR_EVENT_TYPE] hass.bus.async_fire(event_type, data[ATTR_EVENT_DATA], EventOrigin.remote, context=context) return empty_okay_response(headers=headers) if webhook_type == WEBHOOK_TYPE_RENDER_TEMPLATE: resp = {} for key, item in data.items(): try: tpl = item[ATTR_TEMPLATE] attach(hass, tpl) resp[key] = tpl.async_render(item.get(ATTR_TEMPLATE_VARIABLES)) # noqa: E722 pylint: disable=broad-except except TemplateError as ex: resp[key] = {"error": str(ex)} return webhook_response(resp, registration=registration, headers=headers) if webhook_type == WEBHOOK_TYPE_UPDATE_LOCATION: see_payload = { ATTR_DEV_ID: registration[ATTR_DEVICE_ID], ATTR_LOCATION_NAME: data.get(ATTR_LOCATION_NAME), ATTR_GPS: data.get(ATTR_GPS), ATTR_GPS_ACCURACY: data.get(ATTR_GPS_ACCURACY), ATTR_BATTERY: data.get(ATTR_BATTERY), ATTR_ATTRIBUTES: { ATTR_SPEED: data.get(ATTR_SPEED), ATTR_ALTITUDE: data.get(ATTR_ALTITUDE), ATTR_COURSE: data.get(ATTR_COURSE), ATTR_VERTICAL_ACCURACY: data.get(ATTR_VERTICAL_ACCURACY), } } try: await hass.services.async_call(DT_DOMAIN, DT_SEE, see_payload, blocking=True, context=context) # noqa: E722 pylint: disable=broad-except except (vol.Invalid, ServiceNotFound, Exception) as ex: _LOGGER.error("Error when updating location during mobile_app " "webhook (device name: %s): %s", registration[ATTR_DEVICE_NAME], ex) return empty_okay_response(headers=headers) if webhook_type == WEBHOOK_TYPE_UPDATE_REGISTRATION: new_registration = {**registration, **data} hass.data[DOMAIN][DATA_REGISTRATIONS][webhook_id] = new_registration try: await store.async_save(savable_state(hass)) except HomeAssistantError as ex: _LOGGER.error("Error updating mobile_app registration: %s", ex) return empty_okay_response() return webhook_response(safe_registration(new_registration), registration=registration, headers=headers)