Minwook Je

Posted on Dec 3

AWS IoT

#aws #iot #security

1. Policy

Robot Policy

[보안 (Security)] -> [정책 (Policies)] 선택

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": "iot:Connect",
      "Resource": "arn:aws:iot:*:*:client/${iot:Connection.Thing.ThingName}"
    },
    {
      "Effect": "Allow",
      "Action": ["iot:Publish", "iot:Receive"],
      "Resource": "arn:aws:iot:*:*:topic/$aws/things/${iot:Connection.Thing.ThingName}/*"
    },
    {
      "Effect": "Allow",
      "Action": "iot:Subscribe",
      "Resource": "arn:aws:iot:*:*:topicfilter/$aws/things/${iot:Connection.Thing.ThingName}/*"
    },
    {
      "Effect": "Allow",
      "Action": ["iot:UpdateThingShadow", "iot:GetThingShadow"],
      "Resource": "arn:aws:iot:*:*:thing/${iot:Connection.Thing.ThingName}"
    }
  ]
}

TabletPolicy

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": "iot:Connect",
      "Resource": "arn:aws:iot:*:*:client/*"
    },
    {
      "Effect": "Allow",
      "Action": ["iot:Publish", "iot:Receive", "iot:Subscribe"],
      "Resource": "arn:aws:iot:*:*:topic/$aws/things/*/shadow/*"
    },
    {
      "Effect": "Allow",
      "Action": "iot:Subscribe",
      "Resource": "arn:aws:iot:*:*:topicfilter/$aws/things/*/shadow/*"
    }
  ]
}

2. Thing & Attributes

Robot 종류(HW&SW): Thing Type

하드웨어 모델은 절대 바뀌지 않으니, Type으로 정의

Type_G
Type_G
Type_Tablet_

Store: Thing Group
로봇과 태블릿은 '장소' 기준으로 묶어서 관리, Static Group 사용.

Store_* prefix를 사용하는 group생성

Store_Office
Store_Google
...

NUC: Thing

실제 물리적인 기기 하나하나, Type과 Group에 소속 시킴.

e.g., Robot_001 (Type: G_X, Group: Store_Office)

mac address로 등록하고, type으로 확인하면 된다.

$ cat /sys/class/net/wlp46s0/address

Certificate

Device Certificate (crt)
pub key (key)
private key (key)
Root CA certificate
- CA1 (RSA 2048
- CA2 ECC 256

SW 버전: Device Shadow

SW 버전은 계속 업데이트되면서 변하는 값, 그러니 상태값으로 관리되게한다.

3. Robot Client

# 1. 프로젝트 폴더 생성
mkdir -p ~/robot-agent/certs

# 2. 다운로드 받은 키 파일 3개를 여기로 옮기세요.
# (파일 이름을 헷갈리지 않게 바꾸는 걸 추천합니다)
# 예시:
#   xxxxxxxx-certificate.pem.crt  ->  device_cert.pem
#   xxxxxxxx-private.pem.key      ->  private.key
#   AmazonRootCA1.pem             ->  root_ca.pem

/home/유저명/robot-agent/
  ├── certs/
  │    ├── device_cert.pem
  │    ├── private.key
  │    └── root_ca.pem
  └── test_connect.py  (이제 만들 것)

# pip 업데이트
python3 -m pip install --upgrade pip

# AWS IoT SDK v2 설치 (최신 표준)
python3 -m pip install awsiotsdk

Simple Test

vim ~/robot-agent/test_connect.py

import time
from awscrt import io, mqtt
from awsiot import mqtt_connection_builder

# ============================================================
# [설정] 여기만 본인 환경에 맞춰 수정하세요
# ============================================================
ENDPOINT = "xxxxxxxxxxxx-ats.iot.ap-northeast-2.amazonaws.com" # Console > 설정 > 엔드포인트 복사
CLIENT_ID = "Robot_Naver_01" # 아까 등록한 Thing 이름 (또는 MAC주소)

# 인증서 경로 (폴더 위치 정확히!)
PATH_TO_CERT = "./certs/device_cert.pem"
PATH_TO_KEY = "./certs/private.key"
PATH_TO_ROOT = "./certs/root_ca.pem"
# ============================================================

def on_connection_interrupted(connection, error, **kwargs):
    print(f"연결 끊김! 에러: {error}")

def on_connection_resumed(connection, return_code, session_present, **kwargs):
    print(f"연결 복구됨! (Return Code: {return_code})")

if __name__ == '__main__':
    # 1. 이벤트 루프 생성
    event_loop_group = io.EventLoopGroup(1)
    host_resolver = io.DefaultHostResolver(event_loop_group)
    client_bootstrap = io.ClientBootstrap(event_loop_group, host_resolver)

    # 2. MQTT 연결 빌더 설정
    mqtt_connection = mqtt_connection_builder.mtls_from_path(
        endpoint=ENDPOINT,
        cert_filepath=PATH_TO_CERT,
        pri_key_filepath=PATH_TO_KEY,
        client_bootstrap=client_bootstrap,
        ca_filepath=PATH_TO_ROOT,
        on_connection_interrupted=on_connection_interrupted,
        on_connection_resumed=on_connection_resumed,
        client_id=CLIENT_ID,
        clean_session=False,
        keep_alive_secs=30
    )

    print(f"AWS IoT Core ({ENDPOINT}) 에 연결 시도 중...")

    # 3. 연결 수행
    connect_future = mqtt_connection.connect()

    try:
        connect_future.result(timeout=10) # 10초 대기
        print("✅ 연결 성공! (Connected)")

        # 4. 메시지 한 번 쏴보기
        topic = f"test/topic/{CLIENT_ID}"
        print(f"메시지 발행 중... (Topic: {topic})")
        mqtt_connection.publish(
            topic=topic,
            payload="Hello AWS! I am Robot.",
            qos=mqtt.QoS.AT_LEAST_ONCE
        )
        print("메시지 전송 완료.")

        # 5초 대기 후 종료
        time.sleep(5)

        disconnect_future = mqtt_connection.disconnect()
        disconnect_future.result()
        print("연결 종료.")

    except Exception as e:
        print(f"❌ 연결 실패! 설정을 확인하세요.\n에러 내용: {e}")

cd ~/robot-agent
python3 test_connect.py

AWS IoT Core (xxxxx.amazonaws.com) 에 연결 시도 중...
✅ 연결 성공! (Connected)
메시지 발행 중...
메시지 전송 완료.
연결 종료.

scp -r c4_0f_08_a4_33_0f leo@192.168.0.3:/home/leo/

DEV Community