POSA_LEAKSMS/raspberry_pi4/test_connection.py

#!/usr/bin/env python3
"""서버 연결 테스트 스크립트 — Pi 4 에서 실행.
   비밀값(API_URL/API_KEY)은 config_local.py 에서 로드한다 (config 경유)."""

import hashlib
import hmac
import sys
import time
import subprocess

from config import API_URL, API_KEY, DEVICE_ID, DEVICE_LOCATION, SENSORS

signed_ok = False  # 서명 POST 200 여부 — setup.sh 가 이 exit 코드로 하드닝을 게이트한다

# 이 기기에 설정된 첫 센서로 테스트한다 — sensor_id 를 하드코딩하면(예: 1)
# 다른 지점 노드에서 테스트할 때 그 지점의 leak_sensor_status 를 오염시킨다.
TEST_SENSOR_ID = SENSORS[0]["id"]
TEST_SENSOR_NAME = SENSORS[0]["name"]

# API_URL 에서 베이스 경로 추출 (.../api/sensor_data.php → .../api)
API_BASE = API_URL.rsplit("/", 1)[0]

print("=" * 50)
print(" 서버 연결 테스트 (Pi 4)")
print("=" * 50)
print(f" API_BASE = {API_BASE}")

# ── 1단계: curl로 test.php 호출 ──
print("\n[1/3] curl로 test.php 연결 확인...")
try:
    result = subprocess.run(
        ["curl", "-s", "-o", "/dev/null", "-w", "%{http_code}",
         "-A", "LeakSensor/2.0",
         "-X", "POST", f"{API_BASE}/test.php"],
        capture_output=True, text=True, timeout=10
    )
    code = result.stdout.strip()
    print(f"  HTTP 응답코드: {code}")
    if code == "200":
        print("  → 성공! 서버 연결 정상")
    elif code == "403":
        print("  → 403 Forbidden: 서버/호스팅 자체가 차단 중")
        print("  → 호스팅 설정 또는 상위 .htaccess 확인 필요")
    else:
        print(f"  → 예상 외 응답: {code}")
except Exception as e:
    print(f"  → curl 실패: {e}")

# ── 2단계: curl로 sensor_data.php 호출 ──
print("\n[2/3] curl로 sensor_data.php 연결 확인...")
try:
    result = subprocess.run(
        ["curl", "-s", "-w", "\n%{http_code}",
         "-A", "LeakSensor/2.0",
         "-X", "POST", API_URL,
         "-H", "Content-Type: application/json",
         "-d", f'{{"device_id":"test","sensor_id":{TEST_SENSOR_ID},"event_type":"startup","is_leak":0,"timestamp":1234567890}}'],
        capture_output=True, text=True, timeout=10
    )
    lines = result.stdout.strip().split("\n")
    code = lines[-1] if lines else "?"
    body = "\n".join(lines[:-1])
    print(f"  HTTP 응답코드: {code}")
    print(f"  응답 내용: {body[:200]}")
except Exception as e:
    print(f"  → curl 실패: {e}")

# ── 3단계: Python requests로 서명 포함 호출 ──
print("\n[3/3] Python requests로 sensor_data.php 호출...")
try:
    import requests

    # raw-body X-Signature HMAC (서버 verify_signature_raw 와 동일 — leak_sensor.leak_event_body
    # 와 같은 고정 필드 순서로 본문 문자열을 만들고 그 바이트에 서명한 뒤 그대로 POST 한다).
    ts = int(time.time())
    body = (
        '{'
        f'"device_id":"{DEVICE_ID}",'
        f'"device_location":"{DEVICE_LOCATION}",'
        f'"sensor_id":{TEST_SENSOR_ID},'
        f'"sensor_name":"{TEST_SENSOR_NAME} (연결테스트)",'
        f'"event_type":"startup",'
        f'"is_leak":0,'
        f'"timestamp":{ts}'
        '}'
    )
    body_bytes = body.encode("utf-8")
    signature = hmac.new(API_KEY.encode("utf-8"), body_bytes, hashlib.sha256).hexdigest()

    resp = requests.post(
        API_URL,
        data=body_bytes,
        timeout=10,
        headers={
            "Content-Type": "application/json",
            "X-Device-Id": DEVICE_ID,
            "X-Signature": signature,
            "User-Agent": "LeakSensor/2.0",
        }
    )

    print(f"  HTTP 응답코드: {resp.status_code}")
    print(f"  응답 내용: {resp.text[:200]}")

    if resp.status_code == 200:
        signed_ok = True
        print("  → 성공! 모든 연결 정상")
    elif resp.status_code == 403:
        print("  → 403: requests 라이브러리가 차단됨 또는 서명 불일치")
        print(f"  → User-Agent: {resp.request.headers.get('User-Agent', '?')}")
    else:
        print(f"  → 응답코드: {resp.status_code}")

except ImportError:
    print("  → requests 미설치. install.sh 실행 또는 sudo apt install python3-requests")
except Exception as e:
    print(f"  → 실패: {e}")

print("\n" + "=" * 50)
print(" 테스트 완료" + ("" if signed_ok else " — 서명 POST 실패(위 로그 확인)"))
print("=" * 50)
sys.exit(0 if signed_ok else 1)