LG ThinQ 스마트솔루션 API: AWS IoT MQTT Broker (1)

Utilize LG ThinQ Smart Solution API: AWS IoT Core MQTT (1)

LG ThinQ 스마트솔루션 API 활용법에 대한 내용을 시리즈로 발행해볼까 한다

첫번째 글: LG ThinQ(씽큐) 플랫폼 API 공개 및 OpenAPI 사용해보기 (스마트솔루션 API)

LG ThinQ(씽큐) 플랫폼 API 공개 및 OpenAPI 사용해보기 (스마트솔루션 API)

간단한 사용법은 알았으니 이제 심화 과정으로 들어갈 차례! (심화과정이라고 하지만 중급 개발자 이상에게는 별 볼일 없는 수준의 내용일 수 있는 점 주의 ㅋㅋ)

---

LG ThinQ IoT 플랫폼은 아마존 웹 서비스(AWS)의 IoT Core MQTT 브로커를 사용해 MQTT 구독/발행 메커니즘을 통해 디바이스간 상태 및 명령 전송 기능을 구행한다

따라서 ThinQ 디바이스에서 발행하는 push notification이나 event 발생 정보 등을 클라이언트 입장에서 받기 위해서는 AWS IoT Core MQTT에 접속해 특정 Topic을 구독해야 하는데, 이를 위해서는 MQTT 브로커의 주소(URL) 및 구독할 토픽(topic) 정보를 알아내야 한다

AWS IoT Core의 보안 정책을 따라야 하기 때문에 <클라이언트 장비(PC 등)에서 생성한 SSH 인증서> 및 <이를 토대로 만든 CSR(Certificate Signing Request) 인증서>가 필요하며, 이를ㅡ API를 통해 POST하여 AWS로부터 인증서를 발급>받아 이를 ㅡMQTT Client의 TLS/SSL 보안 인증에 사용>해야 하는 등 꽤 복잡한 프로세스를 거쳐야 한다

본 포스팅에서는 초심자도 별다른 어려움 없이 쉽게 따라할 수 있게 최대한 이론적인 내용은 배제한 채 Python을 통해 쉽게 LG ThinQ AWS IoT Core MQTT 브로커에 접속하는 과정을 소개하도록 한다

0. 사전 준비

인증서 관련 작업에 익숙하지 않은 초급자들을 위해 OpenSSL을 통한 인증서 생성도 Python으로 다뤄보도록 한다

아래 3개 패키지를 pip를 통해 설치해주자 (requests, pyOpenSSL, paho-mqtt)

$ python -m pip install requests
$ python -m pip install pyOpenSSL
$ python -m pip install paho-mqtt==1.6.1

※ paho-mqtt 2.x 버전은 내가 아직 써보지 못해 익숙한 1.6.1 버전 기준으로 글을 작성 (혹시 2.x 버전으로 구동하는 코드가 필요하다면 댓글로 요청 바람)

상기 링크에서와 같이 ThinQ 스마트솔루션 PAT(Personal Access Token)을 발급받은 뒤, API 호출 시 사용할 request header를 생성해주는 함수를 선언해주자

import json
import random
import requests

base_url: str = "https://api-kic.lgthinq.com"
pat_token: str = "thinqpat_xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx"  # 발급받은 PAT로 대체
client_id: str = "yogyui-thinq-api-tester"  # 적절한 이름으로 변경
api_key: str = "v6GFvkweNo7DK7yD3ylIZ9w52aKBU0eJ7wLXkSR3"

def generate_random_string(length: int) -> str:
    characters = "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789"
    result = ''
    for i in range(length):
        result += characters[random.randint(0, len(characters) - 1)]
    return result

def generate_request_header() -> dict:
    return {
        "Authorization": "Bearer " + pat_token,
        "x-message-id": generate_random_string(22),
        "x-country": "KR",
        "x-client-id": client_id,
        "x-api-key": api_key
    }

1. MQTT Broker 접속 주소(URL) 쿼리하기 (Query Domain Names)

아래와 같이 간단하게 python requests를 사용해 GET 호출 후 3개의 URL을 변수로 저장해두도록 하자

※ 헤더에 x-service-phase 항목이 필수로 들어가야 하며, 아래 코드와 같이 헤더 딕셔너리를 update해주면 된다

def query_domain_names() -> tuple:
    url = base_url + "/route"
    headers = generate_request_header()
    headers.update({"x-service-phase": "OP"})
    res = requests.get(url, headers=headers)
    if not res.status_code == 200:
        raise ValueError(f"query_domain_names::Request Failed ({res.status_code}): {res.reason}")
    obj = json.loads(res.content.decode())
    print(f"query_domain_names::success ({obj.get('timestamp')})")
    response = obj.get("response")
    return response.get("apiServer"), response.get("mqttServer"), response.get("webSocketServer")

아래와 같이 함수를 호출하면 된다

apiServer, mqttServer, webSocketServer = query_domain_names()

URL 문자열이 기록된 각 변수를 개별로 확인해보자

이 중 mqttServer의 값이 이 후 mqtt client로 접속할 주소가 된다

※ 개인별로 부여된 mqtt 및 web socket server 주소는 다르기 때문에 외부로 유출되지 않도록 주의!

2. Client 등록하기 (Register Client)

이제 AWS IoT Core에 unique한 클라이언트 ID를 등록해 접속할 준비를 해야 한다

 

클라이언트 등록/해제 함수는 아래와 같이 간단하게 작성할 수 있다 (requests의 POST/DELETE 사용)

def register_client():
    url = base_url + "/client"
    headers = generate_request_header()
    payload = {
        "body": {
            "type": "MQTT",
            "service-code": "SVC202",
            "device-type": "607"
        }
    }
    res = requests.post(url, headers=headers, json=payload)
    if not res.status_code == 200:
        raise ValueError(f"register_client::Request Failed ({res.status_code}): {res.reason}, {res.text}")
    obj = json.loads(res.content.decode())
    print(f"register_client::success ({obj.get('timestamp')})")

def unregister_client():
    url = base_url + "/client"
    headers = generate_request_header()
    payload = {
        "body": {
            "type": "MQTT",
            "service-code": "SVC202"
        }
    }
    res = requests.delete(url, headers=headers, json=payload)
    if not res.status_code == 200:
        raise ValueError(f"unregister_client::Request Failed ({res.status_code}): {res.reason}, {res.text}")
    obj = json.loads(res.content.decode())
    print(f"unregister_client::success ({obj.get('timestamp')})")

다음과 같이 호출

register_client()
# unregister_client()

2.1. 주의사항

특정 클라이언트 ID를 등록해야만 AWS IoT Core로부터 MQTT 브로커에 접속할 수 있는 인증서를 발급받을 수 있다 

동일한 client_id로 등록을 시도할 경우 아래와 같이 "Already registered Client" 오류 메시지를 보게 된다

만약 개발과정에서 이미 등록한 client ID를 다시 등록하고자 할 경우, 등록을 해제(unregister_client)한 뒤 재시도해야 한다

3. AWS 인증서 발급 (Issue Certificates) 및 MQTT 토픽 확인

※ 이 과정은 위의 "2. Client 등록하기"에서와 같이 클라이언트 등록을 한 상태에서만 정상적으로 진행된다

POST 시 "디바이스에서 자체 생성한 Private Key 기반의 CSR 데이터"가 필요하다고 명시되어 있다

TLS/SSL 보안 인증에 익숙하지 않다면 멘붕에 빠질 만큼 무성의한 설명이 아닐 수 없다 ㅋㅋ

 

openssl 등의 툴로 작업해도 되지만, 초심자들을 위해 python으로 한 큐에 공개키-비밀키 생성 및 이를 바탕으로 CSR 인증서까지 만들어주는 코드를 소개하도록 한다

3.1. CSR(Certificate Signing Request) 인증서 만들기

openssl 등을 이용해 별도로 생성해둔 공개키-비밀키 쌍이 있고 CSR을 생성해뒀다면 이 단계는 스킵!

import os
from OpenSSL import crypto

def generate_csr_certificate():
    privkey_pem_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "privkey.pem")
    csr_pem_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "csr.pem")
    if not os.path.isfile(csr_pem_path):
        keypair = crypto.PKey()
        keypair.generate_key(crypto.TYPE_RSA, 2048)
        privkey_pem = crypto.dump_privatekey(crypto.FILETYPE_PEM, keypair).decode(encoding='utf-8')
        with open(privkey_pem_path, 'w') as fp:
            fp.write(privkey_pem)
        
        req = crypto.X509Req()
        req.get_subject().CN = "AWS IoT Certificate"
        req.get_subject().O = "Amazon"
        req.set_pubkey(keypair)
        req.sign(keypair, "sha256")
        csr = crypto.dump_certificate_request(crypto.FILETYPE_PEM, req).decode(encoding='utf-8')
        with open(csr_pem_path, 'w') as fp:
            fp.write(csr)

• pyOpenSSL의 crypto 모듈을 사용해 RSA 알고리즘으로 공개키-비밀키 쌍을 생성
• 비밀키 인증서를 privkey.pem 파일로 로컬에 저장 (AWS ioT Core MQTT 브로커 접속 시 필요)
• 생성된 공개키를 기반으로 X509 PKI(Public Key Infrastructure)로 CSR 인증서 생성
• CSR 인증서를 csr.pem 파일로 로컬에 저장 (AWS ioT Core MQTT 브로커 접속 시 필요)
• csr.pem 파일은 한번만 생성하면 되므로 파일 존재 유무를 판단
• 공개키는 굳이 파일로 저장할 필요는 없다

아래 함수를 호출하면 python 스크립트가 존재하는 동일 경로에 "privkey.pem" 파일과 "csr.pem" 파일이 생성되는 것을 확인할 수 있다

generate_csr_certificate()

인증서 파일을 텍스트 에디터로 열어보면 내부 내용은 다음과 같다

• 개인키 인증서는 "-----BEGIN PRIVATE KEY-----" 문자열로 시작
• 서명 요청 인증서는 "-----BEGIN CERTIFICATE REQUEST-----" 문자열로 시작

privkey.pem 파일 내부 내용

csr.pem 파일 내부 내용

※ 두 파일은 보안 정책상 외부로 유출되지 않도록 유의한다

3.2. API 호출: 인증서 발급 및 MQTT Topic 가져오기

3.1.에서 생성한 CSR 인증서로 ThinQ API를 호출하면 AWS IoT Core에 보안 접속하기 위한 인증서를 발급받을 수 있으며, 구독 및 발행 가능한 MQTT 토픽(Topic) 정보를 얻어올 수 있다

인증서는 3.1.에서와 마찬가지로 로컬에 파일(aws_cert.pem)로 저장해두도록 한다

def issue_certificate() -> tuple:
    generate_csr_certificate()
    url = base_url + "/client/certificate"
    csr_pem_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "csr.pem")
    with open(csr_pem_path, 'r') as fp:
        csr = fp.read()
    headers = generate_request_header()
    payload = {
        "body": {
            "service-code": "SVC202",
            "csr": csr
        }
    }
    res = requests.post(url, headers=headers, json=payload)
    if not res.status_code == 200:
        raise ValueError(f"issue_certificate::Request Failed ({res.status_code}): {res.reason}, {res.text}")
    obj = json.loads(res.content.decode())
    print(f"issue_certificate::success ({obj.get('timestamp')})")
    response = obj.get("response")
    result = response.get('result')
    cert_pem_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'aws_cert.pem')
    with open(cert_pem_path, 'w') as fp:
        fp.write(result.get('certificatePem'))
    return result.get('subscriptions'), result.get('publications')

함수 내부에 generate_csr_certifiacte() 함수를 넣어뒀으므로 사용자는 issue_certificate() 함수만 호출하면 개인키, CSR 인증서, AWS 인증서 3개를 한꺼번에 발급받고 로컬에 저장할 수 있다

subscriptions, publications = issue_certificate()

인증서 파일을 텍스트 에디터로 열어보면 내부 내용은 다음과 같다

• "-----BEGIN CERTIFICATE-----" 문자열로 시작
• AWS ioT Core MQTT 브로커 접속 시 필요하며, 마찬가지로 외부로 유출되지 않도록 유의한다

aws_cert.pem 파일 내부 내용

구독 가능한 토픽들(subscriptions) 변수와 발행 가능한 토픽들(publications) 변수를 확인해보면 아래와 같다 (2. 클라이언트 등록 시 사용한 request header의 x-client-id 값이 토픽에 적용된 것을 알 수 있다)

• 구독 가능 토픽
  • app/clients/{client_id}/push
  • app/clients/{client_id}/inbox
• 발생 가능 토픽
  • app/clients/{client-id}/outbox

MQTT 클라이언트로 해당 토픽 구독을 하면 푸시/이벤트 등록한 디바이스에서 발생하는 푸시 알림이나 이벤트 발생 등에 대한 정보를 받아볼 수 있다

3.3. AWS Root CA(Certificate Authority) 인증서 가져오기

마지막으로 MQTT Client로 접속하기 위해 AWS의 Root CA를 얻어와 로컬에 저장해두도록 하자

Root CA 인증서는 공개키 기반이므로 외부로 공개되어 있으며, 아래 함수를 호출해 저장해도 되고, 블로그에 올려둔 첨부파일을 다운로드받아도 무방하다

이미 AWS의 Root CA 인증서를 가지고 있다면 이 단계는 스킵! (혹은 첨부파일 다운로드)

def get_aws_root_ca_cert():
    rootca_pem_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'aws_root_ca.pem')
    if not os.path.isfile(rootca_pem_path):
        url = 'https://www.amazontrust.com/repository/AmazonRootCA1.pem'
        res = requests.get(url)
        if not res.status_code == 200:
            raise ValueError(f"get_aws_root_ca_cert::Request Failed ({res.status_code}): {res.reason}, {res.text}")
        rootca_pem = res.text
        with open(rootca_pem_path, 'w') as fp:
            fp.write(rootca_pem)

get_aws_root_ca_cert()

aws_root_ca.pem

0.00MB

 

참고로 AWS의 Root CA 인증서 내부 내용은 다음과 같다

-----BEGIN CERTIFICATE-----
MIIDQTCCAimgAwIBAgITBmyfz5m/jAo54vB4ikPmljZbyjANBgkqhkiG9w0BAQsF
ADA5MQswCQYDVQQGEwJVUzEPMA0GA1UEChMGQW1hem9uMRkwFwYDVQQDExBBbWF6
b24gUm9vdCBDQSAxMB4XDTE1MDUyNjAwMDAwMFoXDTM4MDExNzAwMDAwMFowOTEL
MAkGA1UEBhMCVVMxDzANBgNVBAoTBkFtYXpvbjEZMBcGA1UEAxMQQW1hem9uIFJv
b3QgQ0EgMTCCASIwDQYJKoZIhvcNAQEBBQADggEPADCCAQoCggEBALJ4gHHKeNXj
ca9HgFB0fW7Y14h29Jlo91ghYPl0hAEvrAIthtOgQ3pOsqTQNroBvo3bSMgHFzZM
9O6II8c+6zf1tRn4SWiw3te5djgdYZ6k/oI2peVKVuRF4fn9tBb6dNqcmzU5L/qw
IFAGbHrQgLKm+a/sRxmPUDgH3KKHOVj4utWp+UhnMJbulHheb4mjUcAwhmahRWa6
VOujw5H5SNz/0egwLX0tdHA114gk957EWW67c4cX8jJGKLhD+rcdqsq08p8kDi1L
93FcXmn/6pUCyziKrlA4b9v7LWIbxcceVOF34GfID5yHI9Y/QCB/IIDEgEw+OyQm
jgSubJrIqg0CAwEAAaNCMEAwDwYDVR0TAQH/BAUwAwEB/zAOBgNVHQ8BAf8EBAMC
AYYwHQYDVR0OBBYEFIQYzIU07LwMlJQuCFmcx7IQTgoIMA0GCSqGSIb3DQEBCwUA
A4IBAQCY8jdaQZChGsV2USggNiMOruYou6r4lK5IpDB/G/wkjUu0yKGX9rbxenDI
U5PMCCjjmCXPI6T53iHTfIUJrU6adTrCC2qJeHZERxhlbI1Bjjt/msv0tadQ1wUs
N+gDS63pYaACbvXy8MWy7Vu33PqUXHeeE6V/Uq2V8viTO96LXFvKWlJbYK8U90vv
o/ufQJVtMVT8QtPHRh8jrdkPSHCa2XV4cdFyQzR1bldZwgJcJmApzyMZFo6IQ6XU
5MsI+yMRQ+hDKXJioaldXgjUkK642M4UwtBV8ob2xJNDd2ZhwLnoQdeXeGADbkpy
rqXRfboQnoZsG4q5WTP468SQvvG5
-----END CERTIFICATE-----

4. Notice

"2. 클라이언트 등록"에서 사용한 클라이언트 ID(x-client-id)를 해제하면 MQTT 서비스는 더 이상 사용할 수 없으며, 각각의 유니크한 클라이언트 ID별로 AWS의 인증서는 다르게 발급되므로 클라이언트 ID는 자주 바꾸지 않는 것이 좋다 (한 번 고유한 ID를 지정한 다음 계속 사용하는 것이 좋다는 의미)

따라서 앞서 소개한 query_domain_names(), register_client(), issue_certificate(), get_aws_root_ca_cert() 함수들은 ThinQ API 기반 어플리케이션 구동 시 최초에 한번만 구동할 수 있도록 구현하는 것이 바람직하다

(한 번 발급받은 인증서는 등록된 클라이언트가 유지되는 한 AWS IoT MQTT 브로커에 계속 접근 가능)

5. What's up Next?

python의 paho-mqtt 패키지를 사용해 MQTT 클라이언트 객체를 만든 뒤, 앞서 발급받은 인증서들을 사용해 AWS IoT MQTT 브로커에 보안 접속 후 Topic 구독을 통해 디바이스에서 발생하는 푸시 혹은 이벤트 정보들을 받고 파싱하여 사용하는 방법에 대해 알아보도록 한다

블로그에서 다루고 있는 내용들은 Github 저장소에 정리 중이며, 정리가 완료되는 대로 또 다른 글을 통해 알릴 예정