クラウドから IoT 電球を動かす~ AWS Summit Japan 展示の裏側

import boto3
from datetime import datetime, timedelta
import json
from urllib import request, parse
import time
from uuid import uuid4
import base64
import os

DEVICES_TABLE_NAME = 'AlexaSkillSimulatorTable'
DEVICES_SECRETS_TABLE_NAME = 'AlexaSkillSimulatorSecretTable'
AMAZON_API_OAUTH_TOKEN = 'https://api.amazon.com/auth/o2/token'
REDIRECT_URL = os.environ.get('REDIRECT_URL', 'https://example.com/callback')
DESCRIPTION = os.environ.get('DESCRIPTION', 'AWS Alexa Smart Home Sample Device')
ENDPOINT_ID = os.environ.get('ENDPOINT_ID', 'sample-device-001')

dynamodb = boto3.resource("dynamodb")

def utc_timestamp():
    return time.strftime("%Y-%m-%dT%H:%M:%S.00Z", time.gmtime())

def get_alexa_secrets():
    try:
        secret_table = dynamodb.Table(DEVICES_SECRETS_TABLE_NAME)
        response = secret_table.get_item(
            Key={'endpointid': ENDPOINT_ID},
            ConsistentRead=True
        )
        
        if 'Item' not in response:
            raise ValueError(f"Secrets not found for endpointId: {ENDPOINT_ID}")
            
        item = response['Item']
        
        return (
            item.get('alexaclientid', ''),
            item.get('alexasecretid', '')
        )
        
    except Exception as e:
        raise ValueError("Failed to retrieve Alexa secrets")

def get_access_token(grant_code):
    alexa_client_id, alexa_secret_id = get_alexa_secrets()
    auth = base64.b64encode(f"{alexa_client_id.strip()}:{alexa_secret_id}".encode()).decode()
    headers = {
        "Content-Type": "application/x-www-form-urlencoded; charset=utf-8",
        "Authorization": f"Basic {auth}"
    }
    data = {
        "grant_type": "authorization_code",
        "code": grant_code,
        "redirect_uri": REDIRECT_URL
    }
    
    try:
        # セキュリティ: HTTPSスキームのみ許可
        if not AMAZON_API_OAUTH_TOKEN.startswith('https://'):
            raise ValueError("Only HTTPS URLs are allowed")
            
        req = request.Request(
            AMAZON_API_OAUTH_TOKEN,
            data=parse.urlencode(data).encode(),
            headers=headers,
            method="POST"
        )
        
        with request.urlopen(req) as res:
            if res.status != 200:
                raise Exception(f"LWA Error: {res.read().decode()}")
            return json.loads(res.read())
            
    except Exception as e:
        raise ValueError("Token exchange failed")

def lambda_handler(event, context):
    header = event["directive"]["header"]

    if header["namespace"] == "Alexa.Discovery" and header["name"] == "Discover":
        return discover_device(header, event["directive"]["payload"])
    elif header["namespace"] == "Alexa.PowerController":
        return handle_switch_request(header, event["directive"])
    elif header["namespace"] == "Alexa.EndpointHealth":
        return handle_health_request(header, event["directive"])
    elif header["name"] == "AcceptGrant":
        return handle_accept_grant_request(header, event["directive"]["payload"])
    elif header["namespace"] == "Alexa" and header["name"] == "ReportState":  
        return handle_report_state(header, event["directive"])  
    return None

def handle_switch_request(header, directive):
    endpoint_id = directive["endpoint"]["endpointId"]
    request_token = directive["endpoint"]["scope"]["token"]
    
    # ディレクティブに基づいて電源状態を決定
    if header["name"] == "TurnOn":
        power_state = "ON"
    elif header["name"] == "TurnOff":
        power_state = "OFF"
    else:
        power_state = "OFF"  # デフォルト

    try:
        table = dynamodb.Table(DEVICES_TABLE_NAME)
        table.update_item(
            Key={"endpointId": endpoint_id},
            UpdateExpression="SET power_state = :val",
            ExpressionAttributeValues={":val": power_state}
        )
    except Exception:
        pass

    response = {
        "event": {
            "header": {
                "namespace": "Alexa",
                "name": "Response",
                "messageId": header["messageId"] + "-response",
                "payloadVersion": "3"
            },
            "endpoint": {
                "scope": {"type": "BearerToken", "token": request_token},
                "endpointId": endpoint_id
            },
            "payload": {}
        },
        "context": {
            "properties": [{
                "namespace": "Alexa.PowerController",
                "name": "powerState",
                "value": power_state,
                "timeOfSample": utc_timestamp(),
                "uncertaintyInMilliseconds": 0
            }]
        }
    }
    
    return response

def handle_report_state(header, directive):
    endpoint_id = directive["endpoint"]["endpointId"]
    request_token = directive["endpoint"]["scope"]["token"]
    
    try:
        table = dynamodb.Table(DEVICES_TABLE_NAME)
        response = table.get_item(Key={"endpointId": endpoint_id})
        item = response.get("Item", {})
    except Exception as e:
        # DynamoDBエラー時はエラーレスポンスを返す
        return {
            "event": {
                "header": {
                    "namespace": "Alexa",
                    "name": "ErrorResponse",
                    "messageId": header["messageId"] + "-response",
                    "payloadVersion": "3"
                },
                "endpoint": {
                    "endpointId": endpoint_id
                },
                "payload": {
                    "type": "ENDPOINT_UNREACHABLE",
                    "message": "Device is currently unreachable"
                }
            }
        }

    return {
        "event": {
            "header": {
                "namespace": "Alexa",
                "name": "StateReport",
                "messageId": header["messageId"] + "-response",
                "payloadVersion": "3"
            },
            "endpoint": {
                "scope": {"type": "BearerToken", "token": request_token},
                "endpointId": endpoint_id
            },
            "payload": {}
        },
        "context": {
            "properties": [{
                "namespace": "Alexa.PowerController",
                "name": "powerState",
                "value": item.get("power_state", "OFF"),
                "timeOfSample": utc_timestamp(),
                "uncertaintyInMilliseconds": 0
            }]
        }
    }

def handle_health_request(header, directive):
    endpoint_id = directive["endpoint"]["endpointId"]

    return {
        "event": {
            "header": {
                "namespace": "Alexa",
                "name": "Response",
                "messageId": header["messageId"] + "-response",
                "payloadVersion": "3"
            },
            "endpoint": {
                "scope": {"type": "BearerToken", "token": directive["endpoint"]["scope"]["token"]},
                "endpointId": endpoint_id
            },
            "payload": {}
        },
        "context": {
            "properties": [{
                "namespace": "Alexa.EndpointHealth",
                "name": "connectivity",
                "value": {"value": "OK"},
                "timeOfSample": utc_timestamp(),
                "uncertaintyInMilliseconds": 0
            }]
        }
    }

def discover_device(header, payload):
    
    endpoints = []

    display_category = "SWITCH"
    
    endpoints.append({
        "endpointId": ENDPOINT_ID,
        "manufacturerName": "AWS Sample Chaoskitty",
        "friendlyName": ENDPOINT_ID,
        "description": DESCRIPTION,
        "displayCategories": [display_category],
        "cookie": {},
        "capabilities": [
            {
                "type": "AlexaInterface",
                "interface": "Alexa",
                "version": "3"
            },
            {
                "type": "AlexaInterface",
                "interface": "Alexa.PowerController",
                "version": "3",
                "properties": {
                    "supported": [{"name": "powerState"}],
                    "proactivelyReported": True,
                    "retrievable": True
                }
            },
            {
                "type": "AlexaInterface",
                "interface": "Alexa.EndpointHealth",
                "version": "3",
                "properties": {
                    "supported": [{"name": "connectivity"}],
                    "proactivelyReported": True,
                    "retrievable": True
                }
            }
        ]
    })
    
    response_header = header.copy()
    response_header["name"] = "Discover.Response"
    
    return {
        "event": {
            "header": response_header,
            "payload": {"endpoints": endpoints}
        }
    }

def handle_accept_grant_request(header, payload):
    try:
        grant_code = payload["grant"]["code"]
        token = get_access_token(grant_code)
        
        user_token = payload["grantee"]["token"]
        expire_time = datetime.utcnow() + timedelta(seconds=token["expires_in"])
        
        try:
            table = dynamodb.Table(DEVICES_TABLE_NAME)
            table.put_item(
                Item={
                    "endpointId": ENDPOINT_ID,
                    "user_token": user_token,
                    "access_token": token["access_token"],
                    "refresh_token": token["refresh_token"],
                    "expire_time": expire_time.isoformat(timespec='seconds') + 'Z',
                    "power_state": "OFF"
                }
            )
        except Exception:
            pass
        
        return {
            "event": {
                "header": {
                    "namespace": "Alexa.Authorization",
                    "name": "AcceptGrant.Response",
                    "messageId": str(uuid4()),
                    "payloadVersion": "3"
                },
                "payload": {}
            }
        }
        
    except Exception as e:
        return {
            "event": {
                "header": {
                    "namespace": "Alexa.Authorization",
                    "name": "ErrorResponse",
                    "messageId": str(uuid4()),
                    "payloadVersion": "3"
                },
                "payload": {
                    "type": "ACCEPT_GRANT_FAILED",
                    "message": "Grant processing failed"
                }
            }
        }