HA Alexa Lamba with Client Certificate

lambda_function.py

"""
Copyright 2019 Jason Hu <awaregit at gmail.com>
Modified 2020 Matthew Hilton <[email protected]>
Refactor and Modernised 2025 Matthew Hilton <[email protected]>

Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at

    http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
"""

import json
import logging
import os
from http import HTTPStatus
from typing import Any

import urllib3

# Configure debug mode
_debug = bool(os.environ.get("DEBUG"))

# Configure logging with enhanced formatting
_log_format = (
    "%(asctime)s - %(name)s - %(levelname)s - [%(filename)s:%(lineno)d] - %(message)s"
)
_formatter = logging.Formatter(_log_format)

# Console handler
_console_handler = logging.StreamHandler()
_console_handler.setFormatter(_formatter)

# Logger setup
_logger = logging.getLogger("HomeAssistant-SmartHome")
_logger.setLevel(logging.DEBUG if _debug else logging.INFO)
_logger.addHandler(_console_handler)

# Suppress debug logs from urllib3
logging.getLogger("urllib3").setLevel(logging.INFO)

verify_ssl = not bool(os.environ.get("NOT_VERIFY_SSL"))
_logger.debug("SSL verification enabled: %s", verify_ssl)
_http = urllib3.PoolManager(
    cert_reqs='CERT_REQUIRED' if verify_ssl else 'CERT_NONE',
    cert_file='/var/task/cert.pem',
    key_file='/var/task/key.pem',
    timeout=urllib3.Timeout(connect=2.0, read=10.0)
)


def lambda_handler(event: dict[str, Any], context: Any) -> dict[str, Any]:
    """Handle incoming Alexa directive.

    Args:
        event: The Alexa directive event payload
        context: AWS Lambda context object

    Returns:
        Response payload for Alexa

    """
    _logger.info("Processing Alexa request")
    _logger.debug("Event payload: %s", json.dumps(event, indent=2))

    try:
        base_url = _get_base_url()
        directive = _get_directive(event)
        _check_payload_version(directive)
        scope = _get_scope(directive)
        _check_scope_type(scope.get("type"))
        token = _get_token(scope.get("token"))
    except (ValueError, KeyError, json.JSONDecodeError) as e:
        _logger.exception("Error processing request")
        return {
            "event": {
                "payload": {
                    "type": "INVALID_REQUEST",
                    "message": str(e),
                }
            }
        }

    try:
        _logger.info("Sending request to Home Assistant")
        response = _http.request(
            "POST",
            f"{base_url}/api/alexa/smart_home",
            headers={
                "Authorization": f"Bearer {token}",
                "Content-Type": "application/json",
            },
            body=json.dumps(event).encode("utf-8"),
        )
    except urllib3.exceptions.HTTPError:
        _logger.exception("Connection error communicating with Home Assistant")
        return {
            "event": {
                "payload": {
                    "type": "INTERNAL_ERROR",
                    "message": "An unexpected error occurred",
                }
            }
        }
    except Exception:
        _logger.exception("Unexpected error")
        return {
            "event": {
                "payload": {
                    "type": "INTERNAL_ERROR",
                    "message": "An unexpected error occurred",
                }
            }
        }

    _logger.debug("Response status: %s", response.status)

    if response.status >= HTTPStatus.BAD_REQUEST:
        response_text = response.data.decode("utf-8")
        _logger.error(
            "Home Assistant returned error %s: %s", response.status, response_text
        )
        error_type = (
            "INVALID_AUTHORIZATION_CREDENTIAL"
            if response.status in (HTTPStatus.UNAUTHORIZED, HTTPStatus.FORBIDDEN)
            else "INTERNAL_ERROR"
        )
        return {
            "event": {
                "payload": {
                    "type": error_type,
                    "message": response_text,
                }
            }
        }

    response_data = json.loads(response.data.decode("utf-8"))
    _logger.info("Successfully processed Alexa request")
    _logger.debug("Response: %s", json.dumps(response_data, indent=2))
    return response_data


def _get_base_url() -> str:
    base_url = os.environ.get("BASE_URL")
    if base_url is None:
        _logger.error("BASE_URL environment variable not set")
        raise ValueError("BASE_URL environment variable must be set")

    base_url = base_url.rstrip("/")
    _logger.debug("Base URL: %s", base_url)
    return base_url


def _get_directive(event: dict[str, Any]) -> dict[str, Any]:
    directive = event.get("directive")
    if directive is None:
        _logger.error("Malformed request: missing directive")
        raise ValueError("Request missing required directive field")
    return directive


def _check_payload_version(directive: dict[str, Any]) -> None:
    payload_version = directive.get("header", {}).get("payloadVersion")
    if payload_version != "3":
        _logger.error("Unsupported payloadVersion: %s", payload_version)
        raise ValueError(f"Only payloadVersion 3 is supported, got {payload_version}")


def _get_scope(directive: dict[str, Any]) -> dict[str, Any]:
    scope = directive.get("endpoint", {}).get("scope")
    if scope is None:
        # token is in grantee for Linking directive
        scope = directive.get("payload", {}).get("grantee")
    if scope is None:
        # token is in payload for Discovery directive
        scope = directive.get("payload", {}).get("scope")

    if scope is None:
        _logger.error("Malformed request: missing scope/token")
        raise ValueError("Request missing scope in endpoint or payload")
    return scope


def _check_scope_type(scope_type: str | None) -> None:
    if scope_type != "BearerToken":
        _logger.error("Unsupported scope type: %s", scope_type)
        raise ValueError(f"Only BearerToken scope is supported, got {scope_type}")


def _get_token(token: str | None) -> str:
    return os.environ.get("LONG_LIVED_ACCESS_TOKEN")
    if token is None and _debug:
        _logger.debug("Token not found in request, using LONG_LIVED_ACCESS_TOKEN")
        token = os.environ.get("LONG_LIVED_ACCESS_TOKEN")

    if token is None:
        _logger.error("No authentication token available")
        raise ValueError("Authentication token is required")
    return token