POSA_LEAKSMS/raspberry_pi4/test_connection.py
cw2007 a72ea18130 i18n(pi): Raspberry Pi 번들 소스 한글 → 영어 (콘솔 mojibake 해소)
헤드리스 Pi 콘솔/journalctl에서 한글이 깨져 보이는 문제로, Pi 번들(02/03/04)에 들어가는 모든 소스의 사람-대상 텍스트를 영어로 전환. 코드·식별자·설정키·경로·숫자·hex·서명 스킴·f-string 플레이스홀더는 불변.

- raspberry_pi4/, raspberry_pi4_sht30/: 로그·주석·문서·센서명·위치 영어화
- tools/make_deploy_bundles.py: 생성 템플릿(config_local.py/setup.sh/SETUP.md/PI_META 위치) 영어화
- tools/harden_pi.sh, tools/set_network.sh: echo/주석 영어화
- WIRING.html: 영어화 + SHT30 전압 경고 정정 반영
- 서버 번들(01, php/)은 한글 유지(브라우저 대시보드, Pi에서 안 깨짐)
- 전 파이썬/셸 문법검사 통과, JSON 본문 서명 무영향
2026-07-10 10:44:13 +09:00

123 lines
4.3 KiB
Python

#!/usr/bin/env python3
"""Server connection test script — run on the Pi 4.
Secrets (API_URL/API_KEY) are loaded from config_local.py (via 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 # whether the signed POST returned 200 — setup.sh gates hardening on this exit code
# Test with the first sensor configured on this device — hardcoding sensor_id (e.g. 1)
# would pollute another site node's leak_sensor_status when testing from that site.
TEST_SENSOR_ID = SENSORS[0]["id"]
TEST_SENSOR_NAME = SENSORS[0]["name"]
# Extract the base path from API_URL (.../api/sensor_data.php → .../api)
API_BASE = API_URL.rsplit("/", 1)[0]
print("=" * 50)
print(" Server connection test (Pi 4)")
print("=" * 50)
print(f" API_BASE = {API_BASE}")
# ── Step 1: call test.php with curl ──
print("\n[1/3] Checking test.php connectivity with curl...")
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 response code: {code}")
if code == "200":
print(" → Success! Server connection OK")
elif code == "403":
print(" → 403 Forbidden: the server/host itself is blocking")
print(" → Check the hosting settings or the upstream .htaccess")
else:
print(f" → Unexpected response: {code}")
except Exception as e:
print(f" → curl failed: {e}")
# ── Step 2: call sensor_data.php with curl ──
print("\n[2/3] Checking sensor_data.php connectivity with curl...")
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 response code: {code}")
print(f" Response body: {body[:200]}")
except Exception as e:
print(f" → curl failed: {e}")
# ── Step 3: signed call via Python requests ──
print("\n[3/3] Calling sensor_data.php with Python requests...")
try:
import requests
# raw-body X-Signature HMAC (same as the server's verify_signature_raw — build the body
# string with the same fixed field order as leak_sensor.leak_event_body, sign those bytes,
# then POST them verbatim).
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} (connection test)",'
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 response code: {resp.status_code}")
print(f" Response body: {resp.text[:200]}")
if resp.status_code == 200:
signed_ok = True
print(" → Success! All connections OK")
elif resp.status_code == 403:
print(" → 403: the requests library is blocked or the signature does not match")
print(f" → User-Agent: {resp.request.headers.get('User-Agent', '?')}")
else:
print(f" → Response code: {resp.status_code}")
except ImportError:
print(" → requests not installed. Run install.sh or sudo apt install python3-requests")
except Exception as e:
print(f" → Failed: {e}")
print("\n" + "=" * 50)
print(" Test complete" + ("" if signed_ok else " — signed POST failed (check the log above)"))
print("=" * 50)
sys.exit(0 if signed_ok else 1)