"""상시 감시·자가복구 (spec §8). 판정 기준 = network_config.json (적용본). 틱 30s(튜너블 10-300), 히스테리시스 2, cooldown 5분, 시간당 4회 한도. 억제: apply busy·CONFIRM_WAIT/seed·apply.service active → 전체 스킵, country-pending → apply.service 에스컬레이션"만" 금지 (전면 정지 금지 — 자가복구 목표 약화). engine.tick() 은 무조건 호출. 정상 틱은 디스크 무기록(RAM 카운터 + 1h heartbeat). gateway ping 은 2틱마다 (interval×2) WARN 로그만.""" import hashlib, json, os, subprocess, threading, time GRACE_S = 120 COOLDOWN_S = 300 HOURLY_MAX = 4 HEARTBEAT_S = 3600 APPLY_SERVICE = "dpworld-network-apply.service" COMPAT_DPWORLDAPP_MD5 = ("a66c515be8de867e276aa7683c5582f1",) # §9 렌더 계약 검증된 빌드 # 복구 사다리: 체크 이름 → [단계1, 단계2, ...], 각 단계 = argv 목록 (§8 표) _LADDERS = { "wlan_module": [[["systemctl", "start", "dpworld-net-recover.service"]]], "wpa_state": [[["wpa_cli", "-i", "wlan0", "reconfigure"]], [["systemctl", "restart", "wpa_supplicant@wlan0.service"]]], # §3.3-5: apply.service 단독은 gateway/metric-only 변경 침묵 누락 → reload 동반 필수 "addr_route": [[["networkctl", "reconfigure", "wlan0"]], [["systemctl", "start", APPLY_SERVICE], ["networkctl", "reload"]]], } def _addrs_in(ip_brief_out): """M2: `ip -br addr show` 출력 → 정확한 주소 목록 (prefix 제거) — substring 오탐 차단 (예: 설정 192.168.55.5 가 라이브 192.168.55.54/24 에 substring 매칭되던 결함).""" addrs = [] for ln in ip_brief_out.splitlines(): parts = ln.split() addrs.extend(t.split("/")[0] for t in parts[2:]) return addrs def _ladder_for(name): if name in ("eth0", "eth1"): return [[["networkctl", "reconfigure", name]]] # 실패한 iface 그 자체 (§8) return _LADDERS.get(name, [[["systemctl", "start", "dpworld-net-recover.service"]]]) class NetworkWatchdog: def __init__(self, db, engine, journal, net_dir, clock=time.monotonic, runner=None): self.db = db; self.engine = engine; self.journal = journal self.net_dir = net_dir; self.clock = clock self.runner = runner or (lambda argv, t=20: _run(argv, t)) self._fail_streak = {} # check name → 연속 실패 수 self._last_recover = {} # check name → monotonic self._recover_times = [] # 최근 1h 발동 시각 self._critical = False self._grace_until = None # start() 시 설정 self._last_heartbeat = 0.0 self._last_results = {} self._tick_count = 0 self._carrier_state = {} # iface → bool (transition WARN 용) self._ping_state = {} # iface → bool (transition WARN 용) self._down_drift_state = {} # M1: iface → bool (DOWN+주소drift transition WARN 용) self._wpa_query_state = None # C3: wpa_cli 쿼리 성공 여부 (transition WARN 용) self._last_apply_id = None # I1b: 마지막으로 본 apply_id — 변경 시 streak 리셋 self._thread = None # Fix #32: guard _critical / _recover_times — reset_critical() (config route # thread) races the watchdog loop thread that mutates the recovery rate-limit. self._lock = threading.Lock() # ── config (net_config DB key, log_config 전례) ───────── def _cfg(self): cfg = {} try: cfg = self.db.get_config("net_config") or {} except Exception: # noqa: BLE001 pass cfg = cfg if isinstance(cfg, dict) else {} # I2: list 등 이형 → 기본값 (thread 사망 금지) try: interval = max(10, min(300, int(cfg.get("watchdog_interval_s", 30)))) except (TypeError, ValueError): interval = 30 return {"enabled": cfg.get("watchdog_enabled", "on") == "on", "auto_recover": cfg.get("watchdog_auto_recover", "on") == "on", "interval_s": interval} # §8 튜너블 3종 # ── 판정 입력 ──────────────────────────────────────────── def _intent(self): try: from network import renderer with open(os.path.join(self.net_dir, "network_config.json"), encoding="utf-8") as f: return renderer.intent_from_persist(json.load(f)) except (OSError, ValueError): return None def _note_carrier(self, ifc, up): prev = self._carrier_state.get(ifc) self._carrier_state[ifc] = up if prev is not False and not up: # transition → WARN 1회 (§8: 복구 대상 아님) self.journal.event("watchdog", phase="detect", action=f"{ifc}_no_carrier", result="warn", detail={"note": "케이블 미연결 — 복구 안 함"}) def _note_wpa_query(self, ok): """C3: wpa_cli 쿼리 실패 transition WARN 1회 — _note_carrier 패턴.""" prev = self._wpa_query_state self._wpa_query_state = ok if prev is not False and not ok: self.journal.event("watchdog", phase="detect", action="wpa_query_failed", result="warn", detail={"note": "wpa_cli 무응답 — UNKNOWN 처리, 복구 안 함 (C3)"}) def _collect_checks(self): """name → bool(healthy). 이름 = wlan_module/wpa_state/addr_route/eth0/eth1/uplink_routes. 테스트에서 통째 주입.""" out = {} # uplink_routes: 선택 iface 감시 — network_config.json 비의존(Codex #9). from network import netmodel, uplink as _uplink try: uc = self.db.get_config("uplink_config") dev = self.db.get_config("device_config") or {} except Exception: # noqa: BLE001 uc, dev = None, {} up_iface = _uplink.selected_iface(uc if isinstance(uc, dict) else {}) rc_u, routes_u = self.runner(["ip", "route", "show"], 5) if rc_u == 0: # rc!=0 = UNKNOWN → 키 생략(거짓 unhealthy 금지, C3 패턴) if up_iface != "wlan0": it2 = netmodel.intent_from_device(dev) _uplink.attach(it2, dev, {"telemetry_iface": up_iface}, live=_uplink.live_subnets(self.runner)) # codex C4: U3 4번째 호출 site — DHCP on-link skip plan, _sk = _uplink.plan_routes(it2) if plan: out["uplink_routes"] = all( (got := _uplink._installed_route(routes_u, r["dest"])) and got["dev"] == up_iface and got["gateway"] == r["gateway"] for r in plan) else: # wlan0 선택: eth0/eth1 stale /32 가 없어야 healthy (Codex #5) out["uplink_routes"] = not _uplink.stale_routes( _uplink.telemetry_targets(dev), routes_u) it = self._intent() if it is None: return out # ★ {} 아님 — uplink 체크 보존(early-return 앞에서 채움) wifi_configured = bool(netmodel.effective_profiles(it)) if wifi_configured: out["wlan_module"] = os.path.isdir("/sys/module/wlan") and \ os.path.exists("/sys/class/net/wlan0") rc, txt = self.runner(["wpa_cli", "-i", "wlan0", "status"], 5) if rc == 0: self._note_wpa_query(True) out["wpa_state"] = "wpa_state=COMPLETED" in txt else: # C3: 쿼리 실패 = UNKNOWN — 키 생략 (unhealthy 오판 → production wpa flap 금지. # 부팅 순서 race: wpa_supplicant 제어 소켓 준비 전 watchdog 선기동) self._note_wpa_query(False) if it["wlan0"]["mode"] == "static": rc2, addrs = self.runner(["ip", "-br", "addr", "show", "wlan0"], 5) if rc2 != 0: # M-D: transient ip(8) failure → UNKNOWN, omit key (mirror wpa_state C3 pattern) pass else: ok = it["wlan0"]["ip"] in _addrs_in(addrs) # M2: 정확 일치 if ok and netmodel.gateway_set(it["wlan0"]): # §8: IP + 기본라우트 일치 # Fix #13: gate on rc — a transient `ip route show` failure # must not register as route-missing (mirror C3 rc-omit at # ~125). On rc!=0 leave `ok` as the address-only result. rc_rt, routes = self.runner(["ip", "route", "show"], 5) if rc_rt == 0: ok = f"default via {it['wlan0']['gateway']} dev wlan0" in routes out["addr_route"] = ok for ifc in ("eth0", "eth1"): if it[ifc]["mode"] == "static" and it[ifc]["ip"]: rc3, addrs3 = self.runner(["ip", "-br", "addr", "show", ifc], 5) # 케이블 미연결/링크 다운은 복구 대상 아님 (§8). .56 실측: 케이블 없는 eth0 은 # NO-CARRIER 가 아니라 state=DOWN 으로 나옴 — 가드 누락 시 5분마다 무의미 # reconfigure → 시간당 4회 한도 도달 → watchdog CRITICAL 자멸. tokens3 = addrs3.split() state3 = tokens3[1] if len(tokens3) >= 2 else "" if "NO-CARRIER" in addrs3 or state3 == "DOWN": self._note_carrier(ifc, False) # v1.6.0 codex M1: 복구는 여전히 보류(자멸 방지)하되, DOWN iface 가 stale/wrong # IP 를 들고 있으면(intent IP 부재) 가시성만 추가 — transition-only WARN. present = _addrs_in(addrs3) drift = bool(present) and it[ifc]["ip"] not in present prev = self._down_drift_state.get(ifc) self._down_drift_state[ifc] = drift if drift and not prev: self.journal.event("watchdog", phase="detect", action=f"{ifc}_down_addr_drift", result="warn", detail={"expected": it[ifc]["ip"], "actual": present, "state": state3, "note": "link DOWN + 주소 불일치 — 복구 보류(케이블/링크 점검)"}) self._fail_streak[ifc] = 0 # M3: carrier 부재 — streak 동결 아닌 리셋 continue self._note_carrier(ifc, True) out[ifc] = it[ifc]["ip"] in _addrs_in(addrs3) # per-iface 키 + M2 정확 일치 return out def _ping_targets(self): """gateway ping 대상 — 테스트 주입점.""" it = self._intent() if it is None: return [] from network import netmodel return [(ifc, it[ifc]["gateway"]) for ifc in ("wlan0", "eth0", "eth1") if it[ifc]["mode"] == "static" and netmodel.gateway_set(it[ifc])] def _gateway_ping_warn(self): """§8 체크 4: 2틱마다 (interval×2) ping — WARN 로그만, 자동복구 절대 금지 (auth-loop 이력 교훈).""" for ifc, gw in self._ping_targets(): rc, _ = self.runner(["ping", "-c", "1", "-W", "2", "-I", ifc, gw], 5) ok = rc == 0 prev = self._ping_state.get(ifc) self._ping_state[ifc] = ok if prev is not False and not ok: # transition 시에만 기록 (스팸 방지) self.journal.event("watchdog", phase="detect", action="gw_ping", result="warn", detail={"iface": ifc, "gateway": gw, "note": "자동복구 없음 (§8)"}) def _service_active(self, unit): rc, _ = self.runner(["systemctl", "is-active", "--quiet", unit], 5) return rc == 0 def _do_recover(self, name, ladder_idx): if name == "uplink_routes": from network import uplink as _uplink # review U4: country 가 reboot-deferred 보류 중이면 uplink self-heal 건너뜀 — apply_uplink # 가 wpa conf 를 NEW country 로 재렌더 + apply.service/wpa reconfigure 로 보류된 regdomain # 을 라이브 누출하는 것 차단(정상 래더의 country_pending 억제와 대칭). if self.engine.country_pending(): self.journal.event("watchdog", phase="recover", action="skip_apply_uplink", result="warn", detail={"check": name, "reason": "country_pending (§8)"}) return uc = self.db.get_config("uplink_config") up_iface = _uplink.selected_iface(uc if isinstance(uc, dict) else {}) try: res = self.engine.apply_uplink(up_iface) state = (res or {}).get("state", "") if state == "COMMITTED": result = "ok" elif state in ("ROLLED_BACK", "BUSY"): result = "warn" else: # FAILED_CRITICAL, ABORTED, FAILED_VALIDATION, UNKNOWN result = "fail" self.journal.event("watchdog", phase="recover", action="apply_uplink", result=result, detail={"iface": up_iface, "state": state}) except Exception as e: # noqa: BLE001 — 복구 실패가 watchdog thread 죽이면 안 됨 self.journal.event("watchdog", phase="recover", action="apply_uplink", result="fail", detail={"error": str(e)}) return ladder = _ladder_for(name) step = ladder[min(ladder_idx, len(ladder) - 1)] for argv in step: # §8 억제 5: country deferred 보류 중엔 apply.service 에스컬레이션만 금지 if self.engine.country_pending() and APPLY_SERVICE in argv: self.journal.event("watchdog", phase="recover", action="skip_apply_service", result="warn", detail={"check": name, "reason": "country_pending (§8)"}) continue rc, out = self.runner(argv, 30) self.journal.event("watchdog", phase="recover", action=" ".join(argv), result="ok" if rc == 0 else "fail", detail={"check": name, "rc": rc}) # ── 틱 ────────────────────────────────────────────────── def reset_critical(self): """#6: 수동 재무장 — #2 config 라우트의 reset_watchdog_critical 가 호출. critical 잠금 + 1h 발동 이력을 비워 자가복구를 즉시 되살린다.""" with self._lock: # Fix #32: serialize with the watchdog loop's tick mutations self._critical = False self._recover_times = [] self.journal.event("watchdog", phase="reset", action="critical_reset_manual", result="ok", detail={}) def tick_once(self): self.engine.tick() # confirm TTL 안전망 — 억제조건·enabled 와 무관하게 항상 (§6.1) cfg = self._cfg() now = self.clock() # #6: 자동 재무장 — 시간당 한도(HOURLY_MAX)로 _critical 잠긴 뒤 1h 창이 비면 스스로 해제. # early-return 전에 평가해야 critical 잠금 상태에서도 재무장이 동작한다. with self._lock: # Fix #32: serialize rate-limit state with reset_critical() self._recover_times = [t for t in self._recover_times if now - t < 3600] rearmed = self._critical and not self._recover_times if rearmed: self._critical = False if rearmed: self.journal.event("watchdog", phase="rate_limit", action="critical_auto_rearmed", result="ok", detail={}) if not cfg["enabled"] or self._critical: return if self._grace_until is None: self._grace_until = now + GRACE_S if now < self._grace_until: return # I1a: CONFIRM_WAIT 포함 억제 — eth1 확인 대기 중 복구 사다리 침범 금지 # (is_busy_or_confirming 부재 FakeEngine 호환: is_busy 폴백) if getattr(self.engine, "is_busy_or_confirming", self.engine.is_busy)(): return # apply 진행/확인 대기 중 일시정지 (§8) if self._service_active(APPLY_SERVICE) or \ self._service_active("dpworld-network-seed.service"): return # dpworldapp 재시작 직후 정당한 재적용과 경합 금지 (§3.3-4) # I1b: 새 apply 가 지나가면 구 구성 기준의 streak/transition 상태는 무효 — 리셋 status_fn = getattr(self.engine, "status", None) if status_fn is not None: try: aid = status_fn().get("apply_id") except Exception: # noqa: BLE001 — status 실패가 watchdog 틱을 막으면 안 됨 aid = self._last_apply_id if aid != self._last_apply_id: self._last_apply_id = aid self._fail_streak.clear() self._ping_state.clear() self._carrier_state.clear() self._down_drift_state.clear() # Fix #18: else a stale DOWN+drift # warning is silenced for the watchdog's lifetime (transition gate) self._tick_count += 1 results = self._collect_checks() self._last_results = results for name, healthy in results.items(): if healthy: self._fail_streak[name] = 0 continue self._fail_streak[name] = self._fail_streak.get(name, 0) + 1 if self._fail_streak[name] < 2: # 히스테리시스 (§8) continue if not cfg["auto_recover"]: self.journal.event("watchdog", phase="detect", action=name, result="warn", detail={"auto_recover": False}) continue if now - self._last_recover.get(name, -1e9) < COOLDOWN_S: continue with self._lock: # Fix #32: serialize rate-limit state with reset_critical() self._recover_times = [t for t in self._recover_times if now - t < 3600] tripped = len(self._recover_times) >= HOURLY_MAX if tripped: self._critical = True recoveries = len(self._recover_times) if tripped: self.journal.event("watchdog", phase="rate_limit", action="critical_stop", result="fail", detail={"recoveries_last_hour": recoveries}) return ladder_idx = max(0, self._fail_streak[name] - 2) self._do_recover(name, ladder_idx) self._last_recover[name] = now with self._lock: # Fix #32: serialize rate-limit state with reset_critical() self._recover_times.append(now) if self._tick_count % 2 == 0: self._gateway_ping_warn() # 2틱마다 = interval×2 (§8 체크 4) if now - self._last_heartbeat >= HEARTBEAT_S: self._last_heartbeat = now self.journal.event("watchdog", phase="heartbeat", action="tick", result="ok", detail={"checks": {k: bool(v) for k, v in results.items()}}) def _check_fw_md5(self): """§9 펌웨어 drift 감시 — 시작 시 1회: 렌더 계약 검증된 dpworldapp 빌드인지. I6: 1MB 청크 read — MemoryMax=48M 하에서 수십 MB 바이너리 전체 read 금지.""" try: h = hashlib.md5() with open("/usr/bin/dpworldapp", "rb") as f: for chunk in iter(lambda: f.read(1024 * 1024), b""): h.update(chunk) md5 = h.hexdigest() except OSError: return if md5 not in COMPAT_DPWORLDAPP_MD5: self.journal.event("watchdog", phase="startup", action="fw_md5_drift", result="warn", detail={"md5": md5, "note": "렌더 계약 미검증 펌웨어 — golden 재캡처 필요 (§9)"}) def snapshot(self): return {"enabled": self._cfg()["enabled"], "critical": self._critical, "fail_streak": dict(self._fail_streak), "last_results": dict(self._last_results)} def start(self): if self._thread: return self._grace_until = self.clock() + GRACE_S self._check_fw_md5() def loop(): while True: try: self.tick_once() time.sleep(self._cfg()["interval_s"]) except Exception: # noqa: BLE001 — watchdog 은 절대 죽지 않는다 time.sleep(30) # I2: sleep 도 try 안 — _cfg 예외로 thread 사망/tight-loop 금지 self._thread = threading.Thread(target=loop, daemon=True, name="net-watchdog") self._thread.start() def _run(argv, timeout): try: p = subprocess.run(argv, capture_output=True, text=True, timeout=timeout) return p.returncode, p.stdout or "" except (subprocess.SubprocessError, OSError) as e: return 1, str(e)