POSA_LEAKSMS/raspberry_pi4_sht30/test_connection.py
cw2007 73e49fd528 feat: RPi4 누수+온습도 2-모듈 단일 서버 통합 + 보안 하드닝
2번 모듈 신규(raspberry_pi4_sht30/): SHT30 온습도 → cafe24, raw-body X-Signature HMAC, 폐쇄망 오프라인 설치(vendor/install), systemd, CRC/포맷 자가점검. firmware sht30_fw wire 포맷과 byte-parity 검증.

1번 모듈 통합(raspberry_pi4/): 누수+카메라를 동일 서버로. 레거시 sha256(key+json) body-sig → X-Signature HMAC 전환, 사진 canonical=device_id|sensor_id|log_id|timestamp.

서버 통합(php/): api/sensor_data.php 누수 이벤트 수신, api/photo_upload.php 신규(HMAC·finfo·서버생성 파일명·nonce 멱등·sensor 결속), photo.php 인증 사진서빙, dashboard.php 누수 패널+사진, sms_send.php 누수 SMS, config.php 태그/요약.

DB: sql/migration_add_leak.sql(가산·멱등), schema_sht30.sql 통합 스키마.

보안 하드닝(외부 펜테스트 대비, 확정 CRITICAL/HIGH 0): cron_heartbeat api_key fail-closed, setup_mfa 부트스트랩 토큰 post-enrollment 잠금, var/ PHP 실행차단, HSTS.

문서/테스트: SYSTEM_OVERVIEW.md, security_hardening_test.php 갱신. 참고: 사용자 요청으로 세션 외 기존 미커밋 변경도 함께 포함.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-23 16:21:11 +09:00

85 lines
3.2 KiB
Python

#!/usr/bin/env python3
"""서버 연결 테스트 — Pi 4 SHT30(2번) 모듈에서 실행.
wire.py 로 본문을 만들고 raw-body HMAC(X-Signature)을 붙여 실제로 한 번
서명된 POST 를 보낸다. 비밀값(API_URL/API_KEY)은 config_local.py 에서 로드한다.
"""
import time
import subprocess
import requests
from config import (
API_URL, API_KEY, DEVICE_ID, DEVICE_LOCATION,
SENSOR_ID, SENSOR_NAME, APP_VERSION,
)
import wire
API_BASE = API_URL.rsplit("/", 1)[0]
print("=" * 50)
print(" 서버 연결 테스트 (Pi 4 SHT30 / 2번 모듈)")
print("=" * 50)
print(f" API_URL = {API_URL}")
# ── 1단계: curl 로 엔드포인트 도달 확인 (서명 없음 → 403 인증실패가 정상) ──
print("\n[1/2] curl 로 sensor_data.php 도달 확인 (서명 없이 → 인증실패 403 이 정상)...")
try:
result = subprocess.run(
["curl", "-s", "-o", "/dev/null", "-w", "%{http_code}",
"-A", "SHT30Sensor/2.0", "-X", "POST", API_URL,
"-H", "Content-Type: application/json", "-d", "{}"],
capture_output=True, text=True, timeout=10
)
code = result.stdout.strip()
print(f" HTTP 응답코드: {code}")
if code in ("400", "403"):
print(" → 엔드포인트 도달 OK (서명/필드 없이 거부됨 — 정상)")
elif code == "200":
print(" → 200? 서버가 서명 없이 수락 — 서버 설정 점검 필요")
elif code == "404":
print(" → 404: API_URL 경로 확인 필요")
else:
print(f" → 응답: {code} (호스팅/방화벽/.htaccess 점검)")
except Exception as e:
print(f" → curl 실패: {e}")
# ── 2단계: wire.py 로 서명한 실제 측정 POST (startup, 더미값) ──
print("\n[2/2] 서명된 측정 POST (startup, 더미 24.00C/48.00%)...")
try:
ts = int(time.time())
body = wire.sht30_event_body(
DEVICE_ID, DEVICE_LOCATION, SENSOR_ID, SENSOR_NAME,
"startup", ts, 24.0, 48.0, "normal", APP_VERSION,
)
sig = wire.sign_raw(API_KEY, body)
print(f" 본문: {body}")
print(f" X-Signature: {sig[:16]}...")
resp = requests.post(
API_URL, data=body.encode("utf-8"),
headers={
"Content-Type": "application/json",
"X-Signature": sig,
"X-Device-Id": DEVICE_ID,
"User-Agent": "SHT30Sensor/2.0",
},
timeout=10,
)
print(f" HTTP 응답코드: {resp.status_code}")
print(f" 응답 내용: {resp.text[:200]}")
if resp.status_code == 200:
print(" → 성공! 서명 검증 통과, 측정 저장됨 (대시보드 sensor_id=2 확인)")
elif resp.status_code == 403:
print(" → 403 인증 실패: API_KEY 가 서버 config.local.php 와 다르거나 서명 불일치")
elif resp.status_code == 400:
print(" → 400: timestamp 신선도(±600s) 확인 — Pi 시각(NTP) 동기 점검")
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(" 테스트 완료")
print("=" * 50)