feat(packaging): nest protocol/ + ocf/ under smartthings_local, add PyPI packaging
Nest the two library packages under a single import namespace so they can ship as one distribution: protocol/ -> smartthings_local/protocol/ ocf/ -> smartthings_local/ocf/ (git mv, history preserved) - Rewrite all imports protocol.* -> smartthings_local.protocol.*, ocf.* -> smartthings_local.ocf.* across the ocf modules, mqtt_demo/ (bridge, descriptor, samples), and tests. - Add pyproject.toml: dist name `smartthings-local`, hatch-vcs versioning from v* tags, wheel ships only smartthings_local/. - Add .github/workflows/publish.yml: build + PyPI Trusted Publishing on v* tags (OIDC, no stored token). - Force-include protocol/ocf_root_ca.pem via [tool.hatch.build] artifacts: it is tracked but matches .gitignore's *.pem, so hatchling's VCS file selection would drop it — and dtls_session.py loads it at runtime. - Update mqtt_demo Dockerfile COPY and deploy.sh tar allowlist to the single smartthings_local/ package. - .gitignore: build artifacts (_version.py, dist/, *.egg-info/). Validated: pytest tests/ (11 passed), python -m build produces sdist + wheel with the pem bundled, fresh pip install resolves all nested imports with the pem readable from site-packages.
This commit is contained in:
Jack Nagy
@@ -0,0 +1,2 @@
|
||||
"""smartthings-local: local CoAP-over-DTLS (OCF) transport + polling layer
|
||||
for Samsung SmartThings appliances."""
|
||||
@@ -0,0 +1,112 @@
|
||||
"""DTLS-layer liveness via CoAP empty-CON ping + poll-success watchdog.
|
||||
|
||||
Each interval_s the task:
|
||||
1. Sends a CoAP ping. This is fire-and-forget — Samsung's RT-OCF
|
||||
doesn't reliably reply with an RST, so the send itself is the
|
||||
keepalive (it tickles Samsung's observer state). The send only
|
||||
fails if the underlying socket is gone, in which case the failure
|
||||
counts toward fail_threshold.
|
||||
2. Calls liveness_fn() if provided. This is the real half-open
|
||||
detection: PollScheduler exposes last_success_ts, and the bridge
|
||||
wraps it as "did we get a 2.05 in the last 60s". If not, count
|
||||
the tick as a failure even though the ping send succeeded.
|
||||
|
||||
After fail_threshold consecutive failures, fires on_unreachable. First
|
||||
success after a fail streak fires on_reachable. Bridge wires these to
|
||||
MQTT availability.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import threading
|
||||
from typing import Callable, Optional
|
||||
|
||||
from smartthings_local.protocol.dtls_session import DtlsCoapSession
|
||||
|
||||
|
||||
class KeepaliveTask:
|
||||
|
||||
def __init__(self,
|
||||
session: DtlsCoapSession,
|
||||
interval_s: float = 25.0,
|
||||
fail_threshold: int = 3,
|
||||
on_reachable: Optional[Callable[[], None]] = None,
|
||||
on_unreachable: Optional[Callable[[], None]] = None,
|
||||
logger=None,
|
||||
liveness_fn: Optional[Callable[[], bool]] = None):
|
||||
self.session = session
|
||||
self.interval_s = interval_s
|
||||
self.fail_threshold = fail_threshold
|
||||
self.on_reachable = on_reachable
|
||||
self.on_unreachable = on_unreachable
|
||||
self.log = logger
|
||||
self.liveness_fn = liveness_fn
|
||||
|
||||
self._fail_streak = 0
|
||||
self._reachable = True
|
||||
self._ping_count = 0
|
||||
self._ping_fail_count = 0
|
||||
|
||||
@property
|
||||
def ping_count(self) -> int:
|
||||
return self._ping_count
|
||||
|
||||
@property
|
||||
def ping_fail_count(self) -> int:
|
||||
return self._ping_fail_count
|
||||
|
||||
@property
|
||||
def reachable(self) -> bool:
|
||||
return self._reachable
|
||||
|
||||
def run_forever(self, stop: threading.Event) -> None:
|
||||
while not stop.wait(self.interval_s):
|
||||
self._tick()
|
||||
|
||||
def _tick(self) -> None:
|
||||
ok = False
|
||||
try:
|
||||
self.session.ping()
|
||||
ok = True
|
||||
except ConnectionError as e:
|
||||
if self.log: self.log.debug("ping: %s", e)
|
||||
except Exception as e:
|
||||
if self.log: self.log.warning("ping: %s", e)
|
||||
# Real half-open detection: ping sends can succeed against a
|
||||
# silently-wedged peer, but polls won't. If the scheduler
|
||||
# hasn't recorded a 2.05 inside the liveness window, treat
|
||||
# this tick as a failure even though the ping itself went out.
|
||||
if ok and self.liveness_fn is not None:
|
||||
try:
|
||||
alive = bool(self.liveness_fn())
|
||||
except Exception as e:
|
||||
if self.log: self.log.warning("liveness_fn: %s", e)
|
||||
alive = True
|
||||
if not alive:
|
||||
if self.log:
|
||||
self.log.warning("liveness: no successful poll "
|
||||
"in the liveness window")
|
||||
ok = False
|
||||
self._ping_count += 1
|
||||
if ok:
|
||||
if not self._reachable:
|
||||
if self.log:
|
||||
self.log.info("ping recovered after %d fails",
|
||||
self._fail_streak)
|
||||
self._reachable = True
|
||||
if self.on_reachable is not None:
|
||||
try: self.on_reachable()
|
||||
except Exception as e:
|
||||
if self.log: self.log.warning("on_reachable: %s", e)
|
||||
self._fail_streak = 0
|
||||
return
|
||||
self._ping_fail_count += 1
|
||||
self._fail_streak += 1
|
||||
if self._reachable and self._fail_streak >= self.fail_threshold:
|
||||
if self.log:
|
||||
self.log.warning("device unreachable after %d ping failures",
|
||||
self._fail_streak)
|
||||
self._reachable = False
|
||||
if self.on_unreachable is not None:
|
||||
try: self.on_unreachable()
|
||||
except Exception as e:
|
||||
if self.log: self.log.warning("on_unreachable: %s", e)
|
||||
@@ -0,0 +1,50 @@
|
||||
"""Periodic OBSERVE re-subscribe.
|
||||
|
||||
CoAP OBSERVE (RFC 7641) has no built-in TTL, but real-world peers age
|
||||
out observer state on their own schedule — Samsung's RT-OCF is known
|
||||
to silently drop notify delivery during cloud auth blips even though
|
||||
the DTLS session stays healthy. Without a re-subscribe, recovery from
|
||||
such a blip requires a full session reconnect.
|
||||
|
||||
This task derregisters the current observer tokens and re-subscribes
|
||||
every `interval_s`. Cheap (one register CON per path), idempotent
|
||||
(Samsung silently no-ops a register on an already-active token), and
|
||||
resilient — individual subscribe failures are logged but don't abort
|
||||
the task.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import threading
|
||||
from typing import Optional
|
||||
|
||||
from smartthings_local.protocol.dtls_session import DtlsCoapSession
|
||||
|
||||
|
||||
class ObserveRefreshTask:
|
||||
|
||||
def __init__(self,
|
||||
session: DtlsCoapSession,
|
||||
paths,
|
||||
interval_s: float = 6 * 3600.0,
|
||||
logger=None):
|
||||
self.session = session
|
||||
self.paths = [list(p) for p in paths]
|
||||
self.interval_s = interval_s
|
||||
self.log = logger
|
||||
self._refresh_count = 0
|
||||
|
||||
@property
|
||||
def refresh_count(self) -> int:
|
||||
return self._refresh_count
|
||||
|
||||
def run_forever(self, stop: threading.Event) -> None:
|
||||
while not stop.wait(self.interval_s):
|
||||
try:
|
||||
self.session.refresh_observes(self.paths)
|
||||
self._refresh_count += 1
|
||||
if self.log:
|
||||
self.log.info("OBSERVE refresh #%d (%d paths)",
|
||||
self._refresh_count, len(self.paths))
|
||||
except Exception as e:
|
||||
if self.log:
|
||||
self.log.warning("OBSERVE refresh: %s", e)
|
||||
@@ -0,0 +1,313 @@
|
||||
"""Tiered adaptive polling against a DtlsCoapSession.
|
||||
|
||||
Tiers are descriptor-declared (hot/warm/cold + sweep). Per tick:
|
||||
each tier whose deadline has passed polls all its paths sequentially
|
||||
on the shared session, writing into the StateCache. The sweep tier
|
||||
issues one Block2 GET of /device/0 and uses
|
||||
StateCache.index_device_tree to fan its result into many href reps.
|
||||
|
||||
Adaptive cadence: when descriptor.is_active(cache.links) returns True
|
||||
and tier.active_interval_s is set, that tier uses the tighter cadence.
|
||||
If the previous health window saw `active_throttle_threshold` timeouts,
|
||||
the throttle drops back to idle cadence even when active=True — the
|
||||
RT-OCF stack wedges under load and stacking poll attempts only makes
|
||||
it worse.
|
||||
|
||||
Per-tier timeouts: tier.timeout_s overrides the scheduler default. Hot
|
||||
tiers want a tight ceiling (e.g. 2s) so one wedged path can't eat
|
||||
several poll cycles.
|
||||
|
||||
Cooldown on timeout: when a path times out, it's deferred for ~3 of
|
||||
its tier's intervals (clamped 5–60s) via the same _defer_until mechanism
|
||||
used for write_in_progress. Breaks the cluster-cascade where the next
|
||||
tier tick fires immediately after an 8s wedge and reattempts the same
|
||||
stalled path.
|
||||
|
||||
Post-write defer: bridge calls write_in_progress(href) before POSTing
|
||||
a write; the scheduler skips that href for settle_s to avoid Samsung's
|
||||
fetchback-revert bug.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import threading
|
||||
import time
|
||||
from dataclasses import dataclass
|
||||
from typing import Callable, Optional, TYPE_CHECKING
|
||||
|
||||
import cbor2
|
||||
|
||||
from smartthings_local.protocol.dtls_session import DtlsCoapSession, fmt_code
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from .state_cache import StateCache
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class PollTier:
|
||||
name: str
|
||||
interval_s: float
|
||||
paths: tuple[tuple[str, ...], ...]
|
||||
active_interval_s: Optional[float] = None
|
||||
is_sweep: bool = False
|
||||
# Per-tier CoAP request timeout. Falls back to PollScheduler.timeout_s
|
||||
# when None. Hot tiers want a tight ceiling (e.g. 2s) so one wedged
|
||||
# path can't eat several poll cycles; sweep tiers tolerate longer.
|
||||
timeout_s: Optional[float] = None
|
||||
|
||||
|
||||
class PollScheduler:
|
||||
|
||||
def __init__(self,
|
||||
session: DtlsCoapSession,
|
||||
cache: 'StateCache',
|
||||
tiers: list[PollTier],
|
||||
is_active_fn: Optional[Callable[[dict[str, dict]], bool]] = None,
|
||||
logger=None,
|
||||
timeout_s: float = 8.0,
|
||||
active_throttle_timeout_threshold: int = 3):
|
||||
self.session = session
|
||||
self.cache = cache
|
||||
self.tiers = tiers
|
||||
self.is_active_fn = is_active_fn
|
||||
self.log = logger
|
||||
self.timeout_s = timeout_s
|
||||
self.active_throttle_threshold = active_throttle_timeout_threshold
|
||||
|
||||
now = time.monotonic()
|
||||
self._next_due: dict[str, float] = {t.name: now for t in tiers}
|
||||
self._defer_until: dict[str, float] = {}
|
||||
self._defer_lock = threading.Lock()
|
||||
self._poll_count = 0
|
||||
self._poll_error_count = 0
|
||||
self._last_active: Optional[bool] = None
|
||||
self._last_throttled: bool = False
|
||||
# Real-liveness signal for KeepaliveTask: updated on every 2.05
|
||||
# we receive from the wire. Initialized to "now" so the first
|
||||
# keepalive tick after start doesn't fire a false unreachable.
|
||||
self._last_success_ts: float = now
|
||||
|
||||
# Per-window tail-latency tracking. Bridge consumes-and-resets
|
||||
# these via take_window_stats() once per HEALTH_INTERVAL_S.
|
||||
# _window_max_rtt_ms tracks SUCCESSFUL polls only — timeouts go
|
||||
# into _window_timeout_count so the dashboard sees the real tail
|
||||
# instead of an 8000ms wall.
|
||||
self._stats_lock = threading.Lock()
|
||||
self._window_max_rtt_ms = 0.0
|
||||
self._window_slow_count = 0
|
||||
self._window_timeout_count = 0
|
||||
# Snapshot of the previous window's timeout count, used by the
|
||||
# active-window throttle to back off when polls are wedging.
|
||||
self._last_window_timeouts = 0
|
||||
self.slow_threshold_ms = 1000.0
|
||||
|
||||
def write_in_progress(self, href: str, settle_s: float = 4.0) -> None:
|
||||
with self._defer_lock:
|
||||
self._defer_until[href] = time.monotonic() + settle_s
|
||||
|
||||
def run_forever(self, stop: threading.Event) -> None:
|
||||
while not stop.is_set():
|
||||
self._run_due_tiers()
|
||||
sleep_for = max(0.05, min(1.0, self._earliest_deadline() - time.monotonic()))
|
||||
if stop.wait(sleep_for):
|
||||
return
|
||||
|
||||
@property
|
||||
def poll_count(self) -> int:
|
||||
return self._poll_count
|
||||
|
||||
@property
|
||||
def poll_error_count(self) -> int:
|
||||
return self._poll_error_count
|
||||
|
||||
@property
|
||||
def last_success_ts(self) -> float:
|
||||
"""Monotonic timestamp of the most recent 2.05 response from any
|
||||
tier. KeepaliveTask uses this as its half-open-detection signal:
|
||||
if no 2.05 has landed in `liveness_window_s`, the session is
|
||||
wedged regardless of whether ping sends succeed."""
|
||||
return self._last_success_ts
|
||||
|
||||
def take_window_stats(self) -> tuple[float, int, int]:
|
||||
"""Return (max RTT ms over successful polls, slow-poll count,
|
||||
timeout count) seen since the last call, and reset all three.
|
||||
Slow threshold is `self.slow_threshold_ms`. The timeout count
|
||||
is snapshotted into `_last_window_timeouts` for the throttle."""
|
||||
with self._stats_lock:
|
||||
out = (self._window_max_rtt_ms,
|
||||
self._window_slow_count,
|
||||
self._window_timeout_count)
|
||||
self._last_window_timeouts = self._window_timeout_count
|
||||
self._window_max_rtt_ms = 0.0
|
||||
self._window_slow_count = 0
|
||||
self._window_timeout_count = 0
|
||||
return out
|
||||
|
||||
def _record_rtt(self, rtt_ms: float, *, timed_out: bool = False) -> None:
|
||||
with self._stats_lock:
|
||||
if timed_out:
|
||||
self._window_timeout_count += 1
|
||||
return
|
||||
if rtt_ms > self._window_max_rtt_ms:
|
||||
self._window_max_rtt_ms = rtt_ms
|
||||
if rtt_ms >= self.slow_threshold_ms:
|
||||
self._window_slow_count += 1
|
||||
|
||||
def _earliest_deadline(self) -> float:
|
||||
return min(self._next_due.values())
|
||||
|
||||
def _run_due_tiers(self) -> None:
|
||||
now = time.monotonic()
|
||||
active = False
|
||||
if self.is_active_fn is not None:
|
||||
try:
|
||||
active = bool(self.is_active_fn(self.cache.snapshot()))
|
||||
except Exception as e:
|
||||
if self.log: self.log.warning("is_active: %s", e)
|
||||
if active != self._last_active:
|
||||
if self.log and self._last_active is not None:
|
||||
self.log.info("active=%s", active)
|
||||
self._last_active = active
|
||||
# Active-window throttle: if the previous health window saw a
|
||||
# cluster of timeouts (RT-OCF wedging under load), drop back to
|
||||
# idle cadence even when is_active=True. Lets the device breathe
|
||||
# instead of stacking poll attempts on a stalled responder.
|
||||
with self._stats_lock:
|
||||
recent_to = self._last_window_timeouts
|
||||
throttled = (active
|
||||
and recent_to >= self.active_throttle_threshold)
|
||||
if throttled != self._last_throttled:
|
||||
if self.log:
|
||||
if throttled:
|
||||
self.log.warning(
|
||||
"active-throttle ON (%d timeouts last window) — "
|
||||
"using idle cadence", recent_to)
|
||||
else:
|
||||
self.log.info("active-throttle OFF")
|
||||
self._last_throttled = throttled
|
||||
effective_active = active and not throttled
|
||||
for tier in self.tiers:
|
||||
if self._next_due[tier.name] > now:
|
||||
continue
|
||||
interval = (tier.active_interval_s
|
||||
if (effective_active and tier.active_interval_s is not None)
|
||||
else tier.interval_s)
|
||||
self._next_due[tier.name] = now + interval
|
||||
try:
|
||||
if tier.is_sweep:
|
||||
self._do_sweep(tier)
|
||||
else:
|
||||
self._do_tier(tier)
|
||||
except Exception as e:
|
||||
self._poll_error_count += 1
|
||||
if self.log: self.log.warning("tier %s: %s", tier.name, e)
|
||||
|
||||
def _tier_timeout(self, tier: PollTier) -> float:
|
||||
return tier.timeout_s if tier.timeout_s is not None else self.timeout_s
|
||||
|
||||
def _cooldown_for(self, tier: PollTier) -> float:
|
||||
# On timeout, defer the wedged href for ~3 cycles (clamped to a
|
||||
# 5–60s band) so repeated tier ticks don't stack attempts on a
|
||||
# stalled responder. Breaks the cluster-cascade we see in the
|
||||
# Poll Max RTT chart during heavy device use.
|
||||
return max(5.0, min(60.0, tier.interval_s * 3.0))
|
||||
|
||||
def _set_cooldown(self, href: str, cooldown_s: float) -> None:
|
||||
with self._defer_lock:
|
||||
self._defer_until[href] = time.monotonic() + cooldown_s
|
||||
|
||||
def _do_tier(self, tier: PollTier) -> None:
|
||||
timeout = self._tier_timeout(tier)
|
||||
cooldown = self._cooldown_for(tier)
|
||||
for i, path in enumerate(tier.paths):
|
||||
href = '/' + '/'.join(path)
|
||||
with self._defer_lock:
|
||||
if self._defer_until.get(href, 0) > time.monotonic():
|
||||
continue
|
||||
if i > 0:
|
||||
self.session.pace()
|
||||
self._poll_count += 1
|
||||
t0 = time.monotonic()
|
||||
try:
|
||||
code, body = self.session.get(list(path), timeout=timeout)
|
||||
except TimeoutError:
|
||||
self._poll_error_count += 1
|
||||
self._record_rtt(0.0, timed_out=True)
|
||||
self._set_cooldown(href, cooldown)
|
||||
if self.log:
|
||||
self.log.warning("poll %s timeout (cooldown %.0fs)",
|
||||
href, cooldown)
|
||||
return
|
||||
except ConnectionError as e:
|
||||
self._poll_error_count += 1
|
||||
self._record_rtt((time.monotonic() - t0) * 1000.0)
|
||||
if self.log: self.log.debug("poll %s: %s", href, e)
|
||||
return
|
||||
except Exception as e:
|
||||
self._poll_error_count += 1
|
||||
self._record_rtt((time.monotonic() - t0) * 1000.0)
|
||||
if self.log: self.log.warning("poll %s: %s", href, e)
|
||||
return
|
||||
self._record_rtt((time.monotonic() - t0) * 1000.0)
|
||||
if code != 0x45 or not body:
|
||||
self._poll_error_count += 1
|
||||
if self.log: self.log.warning("poll %s -> %s", href, fmt_code(code))
|
||||
continue
|
||||
self._last_success_ts = time.monotonic()
|
||||
try:
|
||||
rep = cbor2.loads(body)
|
||||
except Exception as e:
|
||||
self._poll_error_count += 1
|
||||
if self.log: self.log.warning("poll %s cbor: %s", href, e)
|
||||
continue
|
||||
if isinstance(rep, dict):
|
||||
self.cache.apply_rep(href, rep, source='poll')
|
||||
|
||||
def _do_sweep(self, tier: PollTier) -> None:
|
||||
timeout = self._tier_timeout(tier)
|
||||
cooldown = self._cooldown_for(tier)
|
||||
path = list(tier.paths[0])
|
||||
href = '/' + '/'.join(path)
|
||||
t0 = time.monotonic()
|
||||
self._poll_count += 1
|
||||
try:
|
||||
code, body = self.session.get(path, timeout=timeout)
|
||||
except TimeoutError:
|
||||
self._poll_error_count += 1
|
||||
self._record_rtt(0.0, timed_out=True)
|
||||
self._set_cooldown(href, cooldown)
|
||||
if self.log:
|
||||
self.log.warning("sweep %s timeout (cooldown %.0fs)",
|
||||
path, cooldown)
|
||||
return
|
||||
except ConnectionError as e:
|
||||
self._poll_error_count += 1
|
||||
self._record_rtt((time.monotonic() - t0) * 1000.0)
|
||||
if self.log: self.log.debug("sweep %s: %s", path, e)
|
||||
return
|
||||
except Exception as e:
|
||||
self._poll_error_count += 1
|
||||
self._record_rtt((time.monotonic() - t0) * 1000.0)
|
||||
if self.log: self.log.warning("sweep %s: %s", path, e)
|
||||
return
|
||||
self._record_rtt((time.monotonic() - t0) * 1000.0)
|
||||
if code != 0x45 or not body:
|
||||
self._poll_error_count += 1
|
||||
if self.log: self.log.warning("sweep -> %s", fmt_code(code))
|
||||
return
|
||||
self._last_success_ts = time.monotonic()
|
||||
try:
|
||||
tree = cbor2.loads(body)
|
||||
except Exception as e:
|
||||
self._poll_error_count += 1
|
||||
if self.log: self.log.warning("sweep cbor: %s", e)
|
||||
return
|
||||
indexed = self.cache.index_device_tree(tree)
|
||||
for href, rep in indexed.items():
|
||||
with self._defer_lock:
|
||||
if self._defer_until.get(href, 0) > time.monotonic():
|
||||
continue
|
||||
self.cache.apply_rep(href, rep, source='sweep')
|
||||
if self.log:
|
||||
elapsed_ms = (time.monotonic() - t0) * 1000.0
|
||||
self.log.info("sweep complete (%d links, %.0fms)",
|
||||
len(indexed), elapsed_ms)
|
||||
@@ -0,0 +1,96 @@
|
||||
"""Single source of truth for one appliance's state.
|
||||
|
||||
All writers (OBSERVE notify, poll, seed, optimistic) call apply_rep().
|
||||
A registered on_change callback fires after any apply that mutated the
|
||||
cache, which the bridge wires to its MQTT publish gate.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import threading
|
||||
import time
|
||||
from typing import Callable, Optional, Protocol
|
||||
|
||||
|
||||
class _ObservationHook(Protocol):
|
||||
def on_observation(self, state: dict, href: str, rep: dict) -> None: ...
|
||||
|
||||
|
||||
class StateCache:
|
||||
|
||||
def __init__(self, descriptor: '_ObservationHook'):
|
||||
self.descriptor = descriptor
|
||||
self.links: dict[str, dict] = {}
|
||||
self.last_updated: dict[str, float] = {}
|
||||
self.source: dict[str, str] = {}
|
||||
self.descriptor_state: dict = {}
|
||||
self._on_change: Optional[Callable[[bool, str], None]] = None
|
||||
self._lock = threading.RLock()
|
||||
|
||||
def set_on_change(self, cb: Callable[[bool, str], None]) -> None:
|
||||
self._on_change = cb
|
||||
|
||||
def apply_rep(self, href: str, rep: dict, source: str) -> bool:
|
||||
if not isinstance(rep, dict):
|
||||
return False
|
||||
with self._lock:
|
||||
prior = self.links.get(href)
|
||||
changed = prior != rep
|
||||
self.links[href] = rep
|
||||
self.last_updated[href] = time.time()
|
||||
self.source[href] = source
|
||||
hook = self.descriptor.on_observation
|
||||
if hook is not None:
|
||||
try:
|
||||
hook(self.descriptor_state, href, rep)
|
||||
except Exception:
|
||||
pass
|
||||
if self._on_change is not None:
|
||||
try:
|
||||
self._on_change(changed, source)
|
||||
except Exception:
|
||||
pass
|
||||
return changed
|
||||
|
||||
def apply_optimistic(self, href: str, body: dict) -> bool:
|
||||
if not isinstance(body, dict):
|
||||
return False
|
||||
with self._lock:
|
||||
merged = dict(self.links.get(href) or {})
|
||||
merged.update(body)
|
||||
return self.apply_rep(href, merged, source='optimistic')
|
||||
|
||||
@staticmethod
|
||||
def index_device_tree(device0_body) -> dict[str, dict]:
|
||||
"""Turn a /device/0 CBOR list-of-{href, rep} sweep response into
|
||||
a dict keyed by href. Entry [0] is the device-level rep itself
|
||||
and isn't useful here, so it's skipped.
|
||||
|
||||
Replaces the old standalone sensors.index_links — folded in
|
||||
here because every current and future caller immediately feeds
|
||||
the result into apply_rep on this same cache."""
|
||||
out: dict[str, dict] = {}
|
||||
if not isinstance(device0_body, list):
|
||||
return out
|
||||
for entry in device0_body[1:]:
|
||||
if isinstance(entry, dict) and 'href' in entry:
|
||||
out[entry['href']] = entry.get('rep') or {}
|
||||
return out
|
||||
|
||||
def get(self, href: str) -> Optional[dict]:
|
||||
with self._lock:
|
||||
return self.links.get(href)
|
||||
|
||||
def snapshot(self) -> dict[str, dict]:
|
||||
with self._lock:
|
||||
return dict(self.links)
|
||||
|
||||
def freshness_s(self, href: str) -> Optional[float]:
|
||||
ts = self.last_updated.get(href)
|
||||
return None if ts is None else (time.time() - ts)
|
||||
|
||||
def stalest(self) -> Optional[tuple[str, float]]:
|
||||
with self._lock:
|
||||
if not self.last_updated:
|
||||
return None
|
||||
href = min(self.last_updated, key=self.last_updated.get)
|
||||
return href, time.time() - self.last_updated[href]
|
||||
@@ -0,0 +1,139 @@
|
||||
"""CoAP wire encoding/decoding (RFC 7252 + 7641 + 7959).
|
||||
|
||||
Pure functions — no sockets, no DTLS. Split out of the original
|
||||
coap_dtls.py so protocol/dtls_session.py (the stateful session) and
|
||||
this module (stateless wire format) can be reasoned about and tested
|
||||
independently.
|
||||
"""
|
||||
import struct
|
||||
|
||||
# CoAP option numbers (RFC 7252 + 7641 + 7959)
|
||||
URI_PATH = 11
|
||||
URI_QUERY = 15
|
||||
OBSERVE = 6
|
||||
CONTENT_FORMAT = 12
|
||||
ACCEPT = 17
|
||||
BLOCK2 = 23
|
||||
SIZE2 = 28
|
||||
|
||||
# CoAP message types
|
||||
TYPE_CON = 0
|
||||
TYPE_NON = 1
|
||||
TYPE_ACK = 2
|
||||
TYPE_RST = 3
|
||||
|
||||
# CoAP method codes
|
||||
METHOD_GET = 0x01
|
||||
METHOD_POST = 0x02
|
||||
|
||||
# CoAP content-format value for application/cbor
|
||||
CF_CBOR = b'\x3c'
|
||||
|
||||
# OBSERVE option values (RFC 7641 §2)
|
||||
OBSERVE_REGISTER = b'' # register / refresh
|
||||
OBSERVE_DEREGISTER = bytes([1]) # deregister
|
||||
|
||||
# Block2 SZX=6 → 1024-byte blocks. The largest size Samsung's RT-OCF
|
||||
# will honour and the only one the probes have validated end-to-end.
|
||||
BLOCK_SZX = 6
|
||||
|
||||
|
||||
def _vlen(v):
|
||||
"""Variable-length integer encoder used in option deltas + lengths."""
|
||||
if v < 13: return v, b''
|
||||
if v < 269: return 13, bytes([v - 13])
|
||||
return 14, struct.pack('>H', v - 269)
|
||||
|
||||
|
||||
def encode_options(opts):
|
||||
"""Encode a list of (option_number, value_bytes) tuples."""
|
||||
out = b''
|
||||
prev = 0
|
||||
for n, val in sorted(opts, key=lambda x: x[0]):
|
||||
d, dx = _vlen(n - prev)
|
||||
l, lx = _vlen(len(val))
|
||||
out += bytes([(d << 4) | l]) + dx + lx + val
|
||||
prev = n
|
||||
return out
|
||||
|
||||
|
||||
def parse_coap(data):
|
||||
"""Decode a CoAP datagram. Returns (mtype, code, mid, token,
|
||||
options, payload). options is a list of (num, value_bytes)."""
|
||||
mt = (data[0] >> 4) & 0x03
|
||||
tkl = data[0] & 0x0F
|
||||
code = data[1]
|
||||
mid = int.from_bytes(data[2:4], 'big')
|
||||
tok = data[4:4 + tkl]
|
||||
i = 4 + tkl
|
||||
opts = []
|
||||
prev = 0
|
||||
payload = b''
|
||||
while i < len(data):
|
||||
b = data[i]
|
||||
if b == 0xFF:
|
||||
payload = data[i + 1:]
|
||||
break
|
||||
d_nib, l_nib = b >> 4, b & 0x0F
|
||||
i += 1
|
||||
if d_nib == 13:
|
||||
delta = 13 + data[i]; i += 1
|
||||
elif d_nib == 14:
|
||||
delta = 269 + int.from_bytes(data[i:i + 2], 'big'); i += 2
|
||||
elif d_nib == 15:
|
||||
raise ValueError("reserved option delta nibble 15")
|
||||
else:
|
||||
delta = d_nib
|
||||
if l_nib == 13:
|
||||
length = 13 + data[i]; i += 1
|
||||
elif l_nib == 14:
|
||||
length = 269 + int.from_bytes(data[i:i + 2], 'big'); i += 2
|
||||
elif l_nib == 15:
|
||||
raise ValueError("reserved option length nibble 15")
|
||||
else:
|
||||
length = l_nib
|
||||
num = prev + delta
|
||||
opts.append((num, data[i:i + length]))
|
||||
i += length
|
||||
prev = num
|
||||
return mt, code, mid, tok, opts, payload
|
||||
|
||||
|
||||
def build_coap(mtype, code, mid, token, options, payload=b''):
|
||||
"""Build a CoAP datagram. mtype: CON/NON/ACK/RST. token: bytes (may
|
||||
be empty for ACK). options: list of (num, value_bytes)."""
|
||||
tkl = len(token)
|
||||
hdr = bytes([(1 << 6) | (mtype << 4) | tkl, code,
|
||||
(mid >> 8) & 0xFF, mid & 0xFF])
|
||||
body = hdr + token + encode_options(options)
|
||||
if payload:
|
||||
body += b'\xFF' + payload
|
||||
return body
|
||||
|
||||
|
||||
def block_value(num, more, szx):
|
||||
"""Encode a CoAP Block-N option value."""
|
||||
v = (num << 4) | ((more & 1) << 3) | (szx & 7)
|
||||
if v <= 0xFF: return bytes([v])
|
||||
if v <= 0xFFFF: return struct.pack('>H', v)
|
||||
return struct.pack('>I', v)[1:]
|
||||
|
||||
|
||||
def fmt_code(c):
|
||||
"""0x45 → '2.05', 0x84 → '4.04'. Used in log lines."""
|
||||
return f"{c >> 5}.{c & 0x1F:02d}"
|
||||
|
||||
|
||||
def split_dtls(buf):
|
||||
"""Split a UDP datagram that contains one-or-more DTLS records.
|
||||
OpenSSL sometimes hands the BIO multiple records back-to-back; we
|
||||
must send each as its own UDP datagram or TizenRT drops them."""
|
||||
o, out = 0, []
|
||||
while o + 13 <= len(buf):
|
||||
L = int.from_bytes(buf[o + 11:o + 13], 'big')
|
||||
end = o + 13 + L
|
||||
if end > len(buf):
|
||||
break
|
||||
out.append(buf[o:end])
|
||||
o = end
|
||||
return out
|
||||
@@ -0,0 +1,588 @@
|
||||
"""CoAP-over-DTLS client for Samsung RT-OCF appliances (RFC 7252 + 6347).
|
||||
|
||||
Replaces the TLS-over-TCP transport used in the original dryer bridge.
|
||||
Both the oven (UDP/49154) and the dryer (UDP/49155) speak CoAP-over-DTLS
|
||||
with the ECDHE-ECDSA-AES128-GCM-SHA256 cipher and a client cert.
|
||||
|
||||
Wire-level details that matter (from local-tools/oven-findings.md §17):
|
||||
* DTLS ciphertext MTU must be 1200; otherwise OpenSSL fragments the
|
||||
client cert across two datagrams and TizenRT drops the second.
|
||||
* Samsung's RT-OCF uses ACK+separate-CON for the larger responses.
|
||||
The reader MUST correlate by (token, mid) — not arrival order —
|
||||
or interleaved one-shot / OBSERVE traffic mis-attributes.
|
||||
* Multi-block GET requires the SAME CoAP token across every block
|
||||
of the response ("token-stable Block2"). Fresh-token-per-block
|
||||
is silently dropped by the server.
|
||||
|
||||
Reader thread owns the UDP socket. Callers issue get()/post() and block
|
||||
on a per-token Event the reader signals. OBSERVE notifications are
|
||||
delivered via the on_notification callback.
|
||||
"""
|
||||
import os
|
||||
import socket
|
||||
import threading
|
||||
import time
|
||||
from pathlib import Path
|
||||
|
||||
from OpenSSL import SSL
|
||||
|
||||
from .coap import (
|
||||
URI_PATH, URI_QUERY, OBSERVE, CONTENT_FORMAT, ACCEPT, BLOCK2, SIZE2,
|
||||
TYPE_CON, TYPE_NON, TYPE_ACK, TYPE_RST,
|
||||
METHOD_GET, METHOD_POST, CF_CBOR,
|
||||
OBSERVE_REGISTER, OBSERVE_DEREGISTER, BLOCK_SZX,
|
||||
encode_options, parse_coap, build_coap, block_value, fmt_code,
|
||||
split_dtls as _split_dtls,
|
||||
)
|
||||
import logging
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
_OCF_ROOT_CA = str(Path(__file__).parent / 'ocf_root_ca.pem')
|
||||
|
||||
|
||||
# Diagnostic logging — when DEBUG_BRIDGE=1 in env, the bridge dumps
|
||||
# every received CoAP frame, every /operational/state/vs/0 + /oven/vs/0
|
||||
# + /power/vs/0 + /mode/vs/0-options rep change, the full link tree at
|
||||
# seed time, and the /oic/res directory. Useful for reverse-engineering
|
||||
# new resources and field semantics; otherwise quiet.
|
||||
DEBUG_BRIDGE = os.environ.get('DEBUG_BRIDGE') == '1'
|
||||
|
||||
# Per-block retransmission: send up to this many times before giving up.
|
||||
# Each attempt waits at most _BLOCK_ACK_TIMEOUT seconds (capped by the
|
||||
# overall deadline). Matches RFC 7252 CON retransmit behaviour.
|
||||
_BLOCK_MAX_ATTEMPTS = 3
|
||||
_BLOCK_ACK_TIMEOUT = 4.0
|
||||
|
||||
# Inter-request pacing: minimum seconds between CoAP CON sends on one session.
|
||||
# Samsung's RT-OCF stacks drop requests when hit faster than their firmware
|
||||
# ceiling (dryer ~14 req/s, oven ~8 req/s, dishwasher unknown). 5 req/s
|
||||
# (200 ms) is conservative enough for all tested devices; tune per device
|
||||
# once the ceiling is measured empirically.
|
||||
_DEFAULT_RATE_LIMIT_RPS = 5.0
|
||||
|
||||
|
||||
class DtlsCoapSession:
|
||||
"""Single sustained DTLS-CoAP session.
|
||||
|
||||
Caller drives lifecycle:
|
||||
sess = DtlsCoapSession(host, port, cert, key)
|
||||
sess.connect()
|
||||
sess.start_reader()
|
||||
sess.subscribe([...], on_notification=cb) # OBSERVE
|
||||
code, body = sess.get(['device', '0']) # Block2 fetch
|
||||
code, _ = sess.post(['mode','vs','0'], cbor)
|
||||
sess.close()
|
||||
"""
|
||||
|
||||
HANDSHAKE_TIMEOUT_S = 12.0
|
||||
READER_RECV_TIMEOUT_S = 1.0 # short so stop_event propagates quickly
|
||||
MAX_BLOCKS = 32 # safety bound for Block2 fetches
|
||||
|
||||
def __init__(self, host, port, cert_path, key_path,
|
||||
on_notification=None, mtu=1200,
|
||||
rate_limit_rps: float = _DEFAULT_RATE_LIMIT_RPS):
|
||||
self.host = host
|
||||
self.port = port
|
||||
self.cert_path = str(cert_path)
|
||||
self.key_path = str(key_path)
|
||||
self.on_notification = on_notification # fn(href, payload_bytes)
|
||||
self.mtu = mtu
|
||||
self._min_req_interval = 1.0 / rate_limit_rps
|
||||
|
||||
self.sock = None
|
||||
self.conn = None
|
||||
self.dest = None
|
||||
|
||||
self._send_lock = threading.Lock()
|
||||
# Randomize MID and token counter starting points so reconnects
|
||||
# don't reuse identifiers from previous sessions — Samsung's
|
||||
# RT-OCF appears to remember observer state across DTLS
|
||||
# sessions, and re-registering with a token it still thinks is
|
||||
# active is silently no-ops.
|
||||
self._mid = int.from_bytes(os.urandom(2), 'big')
|
||||
self._tok_counter = int.from_bytes(os.urandom(4), 'big')
|
||||
# OBSERVE tokens are 1-byte (Samsung silently drops TKL>1
|
||||
# OBSERVE registrations). Pick a random starting byte in the
|
||||
# 0x40..0xff range so each session uses fresh values.
|
||||
self._observe_tok_counter = 0x40 + (os.urandom(1)[0] & 0xBF)
|
||||
# token (bytes) → (Event, container_dict)
|
||||
self._pending = {}
|
||||
# token (bytes) → href (str)
|
||||
self._observe_tokens = {}
|
||||
|
||||
self._stop = threading.Event()
|
||||
self._reader_thread = None
|
||||
self._last_send_ts = 0.0
|
||||
|
||||
def pace(self) -> None:
|
||||
"""Sleep only the part of the rate-limit interval not already consumed
|
||||
since the last real send. Uses _stop so session teardown wakes it."""
|
||||
remaining = self._min_req_interval - (time.monotonic() - self._last_send_ts)
|
||||
if remaining > 0:
|
||||
self._stop.wait(remaining)
|
||||
|
||||
# ---- lifecycle ---------------------------------------------------
|
||||
|
||||
def connect(self):
|
||||
"""DTLS handshake. Blocks up to HANDSHAKE_TIMEOUT_S. Raises
|
||||
ConnectionError / TimeoutError on failure."""
|
||||
ctx = SSL.Context(SSL.DTLS_METHOD)
|
||||
|
||||
ctx.load_verify_locations(_OCF_ROOT_CA)
|
||||
ctx.set_verify(SSL.VERIFY_PEER, lambda conn, cert, err, depth, ok: ok)
|
||||
# @SECLEVEL=0 permits SHA-1 in Samsung's server cert chain (AC14K_M
|
||||
# intermediate is SHA-1 signed). This is the only channel that reaches
|
||||
# the OpenSSL instance cryptography bundles — ctypes and cffi bindings
|
||||
# do not expose SSL_CTX_set_security_level on this build.
|
||||
ctx.set_cipher_list(b'ECDHE-ECDSA-AES128-GCM-SHA256:@SECLEVEL=0')
|
||||
ctx.use_certificate_chain_file(self.cert_path)
|
||||
ctx.use_privatekey_file(self.key_path)
|
||||
ctx.check_privatekey()
|
||||
|
||||
conn = SSL.Connection(ctx, None)
|
||||
conn.set_connect_state()
|
||||
conn.set_ciphertext_mtu(self.mtu)
|
||||
|
||||
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
|
||||
sock.settimeout(2.0)
|
||||
dest = (self.host, self.port)
|
||||
|
||||
t0 = time.time()
|
||||
while time.time() - t0 < self.HANDSHAKE_TIMEOUT_S:
|
||||
try:
|
||||
conn.do_handshake()
|
||||
break
|
||||
except SSL.WantReadError:
|
||||
pass
|
||||
except SSL.Error as e:
|
||||
sock.close()
|
||||
raise ConnectionError(f"DTLS handshake error: {e}") from e
|
||||
try:
|
||||
o = conn.bio_read(65535)
|
||||
if o:
|
||||
for r in _split_dtls(o):
|
||||
sock.sendto(r, dest)
|
||||
except SSL.WantReadError:
|
||||
pass
|
||||
try:
|
||||
d, _ = sock.recvfrom(65535)
|
||||
if d:
|
||||
conn.bio_write(d)
|
||||
except socket.timeout:
|
||||
pass
|
||||
time.sleep(0.05)
|
||||
else:
|
||||
sock.close()
|
||||
raise TimeoutError(
|
||||
f"DTLS handshake timeout to {self.host}:{self.port}")
|
||||
|
||||
self.sock = sock
|
||||
self.conn = conn
|
||||
self.dest = dest
|
||||
self._stop.clear()
|
||||
|
||||
def start_reader(self):
|
||||
"""Spawn the reader thread. Must be called after connect()."""
|
||||
if self.sock is None:
|
||||
raise RuntimeError("connect() before start_reader()")
|
||||
t = threading.Thread(target=self._reader_loop,
|
||||
daemon=True, name='dtls-reader')
|
||||
t.start()
|
||||
self._reader_thread = t
|
||||
|
||||
def join(self):
|
||||
"""Block until the reader thread exits (i.e. socket dies)."""
|
||||
if self._reader_thread is not None:
|
||||
self._reader_thread.join()
|
||||
|
||||
def _send_observe_dereg(self, tok, path_segs):
|
||||
"""Send a single OBSERVE deregister GET (Observe option = 1)
|
||||
on the existing token. Best-effort — caller swallows errors."""
|
||||
if self.conn is None:
|
||||
return
|
||||
mid = self._next_mid()
|
||||
opts = [(URI_PATH, s.encode()) for s in path_segs]
|
||||
opts.append((OBSERVE, OBSERVE_DEREGISTER))
|
||||
opts.append((ACCEPT, CF_CBOR))
|
||||
self._send_dgram(
|
||||
build_coap(TYPE_CON, METHOD_GET, mid, tok, opts))
|
||||
|
||||
def close(self):
|
||||
"""Tear down session. Sends best-effort OBSERVE deregisters
|
||||
first so Samsung's RT-OCF cleans up its observer table —
|
||||
without this, the per-cert observer state survives DTLS close
|
||||
and a quick reconnect with the same tokens silently no-ops."""
|
||||
# Send dereg for every active observation while the conn is
|
||||
# still healthy. Tiny sleep lets the records reach the wire
|
||||
# before we shut DTLS down.
|
||||
if self.conn is not None and self._observe_tokens:
|
||||
for tok, href in list(self._observe_tokens.items()):
|
||||
segs = [s for s in href.split('/') if s]
|
||||
try:
|
||||
self._send_observe_dereg(tok, segs)
|
||||
except Exception as e:
|
||||
logger.warning("dereg %s: %s", href, e)
|
||||
time.sleep(0.1)
|
||||
|
||||
self._stop.set()
|
||||
if self.conn is not None:
|
||||
try:
|
||||
self.conn.shutdown()
|
||||
except Exception:
|
||||
pass
|
||||
if self.sock is not None:
|
||||
try:
|
||||
self.sock.close()
|
||||
except Exception:
|
||||
pass
|
||||
for tok, (ev, container) in list(self._pending.items()):
|
||||
container.setdefault('err', 'socket closed')
|
||||
ev.set()
|
||||
self._pending.clear()
|
||||
self._observe_tokens.clear()
|
||||
self.sock = None
|
||||
self.conn = None
|
||||
|
||||
# ---- send / receive plumbing -------------------------------------
|
||||
|
||||
def _next_mid(self):
|
||||
self._mid = (self._mid + 1) & 0xFFFF
|
||||
return self._mid
|
||||
|
||||
def _next_tok(self):
|
||||
self._tok_counter = (self._tok_counter + 1) & 0xFFFFFFFF
|
||||
# 4-byte tokens — fits within tkl=8 cap with headroom and
|
||||
# avoids collisions across long-running OBSERVE subscriptions.
|
||||
return self._tok_counter.to_bytes(4, 'big')
|
||||
|
||||
def _next_observe_tok(self):
|
||||
# Single-byte tokens for OBSERVE registrations. Samsung
|
||||
# RT-OCF accepts these but silently drops TKL=4 OBSERVE
|
||||
# registrations. Counter is randomly seeded per session so
|
||||
# reconnects don't collide with stale observer state Samsung
|
||||
# may still be holding from the previous run.
|
||||
self._observe_tok_counter = (self._observe_tok_counter + 1) & 0xFF
|
||||
# Avoid 0x00 — some CoAP stacks treat an all-zero token as
|
||||
# equivalent to "no token" / empty (TKL=0).
|
||||
if self._observe_tok_counter == 0:
|
||||
self._observe_tok_counter = 1
|
||||
return bytes([self._observe_tok_counter])
|
||||
|
||||
def _send_dgram(self, datagram):
|
||||
"""Send a CoAP datagram. Holds the send lock for the
|
||||
BIO-drain so two writers can't interleave records."""
|
||||
with self._send_lock:
|
||||
if self.conn is None:
|
||||
raise ConnectionError("DTLS session closed")
|
||||
try:
|
||||
self.conn.send(datagram)
|
||||
self._last_send_ts = time.monotonic()
|
||||
while True:
|
||||
o = self.conn.bio_read(65535)
|
||||
if not o:
|
||||
break
|
||||
for r in _split_dtls(o):
|
||||
self.sock.sendto(r, self.dest)
|
||||
except SSL.WantReadError:
|
||||
pass
|
||||
|
||||
def _reader_loop(self):
|
||||
"""Pump UDP socket → DTLS BIO → CoAP parser. Demuxes to pending
|
||||
/ observe handlers. Exits on socket error or stop event."""
|
||||
sock = self.sock
|
||||
conn = self.conn
|
||||
sock.settimeout(self.READER_RECV_TIMEOUT_S)
|
||||
try:
|
||||
while not self._stop.is_set():
|
||||
try:
|
||||
d, _ = sock.recvfrom(65535)
|
||||
except socket.timeout:
|
||||
continue
|
||||
except (OSError, ValueError):
|
||||
return
|
||||
if not d:
|
||||
continue
|
||||
# pyOpenSSL's SSL.Connection is not thread-safe — the
|
||||
# same SSL object must not be touched by multiple
|
||||
# threads concurrently. Drain decrypted records into a
|
||||
# local list under _send_lock so the reader never races
|
||||
# a sender's conn.send()/bio_read(). Dispatch happens
|
||||
# AFTER releasing the lock because _dispatch_coap may
|
||||
# call _send_dgram (auto-ACK for CON frames), which
|
||||
# re-acquires the lock — holding it across dispatch
|
||||
# would deadlock.
|
||||
packets = []
|
||||
exit_reader = False
|
||||
with self._send_lock:
|
||||
try:
|
||||
conn.bio_write(d)
|
||||
except SSL.Error as e:
|
||||
logger.warning("DTLS bio_write: %s", e)
|
||||
return
|
||||
while True:
|
||||
try:
|
||||
pl = conn.recv(65535)
|
||||
except SSL.WantReadError:
|
||||
break
|
||||
except SSL.ZeroReturnError:
|
||||
logger.info("DTLS peer closed connection")
|
||||
exit_reader = True
|
||||
break
|
||||
except SSL.Error as e:
|
||||
logger.warning("DTLS recv: %s", e)
|
||||
exit_reader = True
|
||||
break
|
||||
if not pl:
|
||||
break
|
||||
packets.append(pl)
|
||||
for pl in packets:
|
||||
try:
|
||||
self._dispatch_coap(pl)
|
||||
except Exception as e:
|
||||
logger.warning("dispatch: %s", e)
|
||||
if exit_reader:
|
||||
return
|
||||
finally:
|
||||
# Make sure pending waiters don't hang if the reader dies.
|
||||
for tok, (ev, container) in list(self._pending.items()):
|
||||
container.setdefault('err', 'reader exited')
|
||||
ev.set()
|
||||
|
||||
def _dispatch_coap(self, datagram):
|
||||
try:
|
||||
mt, code, mid, tok, ropts, payload = parse_coap(datagram)
|
||||
except Exception as e:
|
||||
logger.debug("malformed CoAP: %s", e)
|
||||
return
|
||||
|
||||
if DEBUG_BRIDGE:
|
||||
kind = ['CON', 'NON', 'ACK', 'RST'][mt]
|
||||
logger.info("rx %s code=%s mid=%04x tok=%s opts=%d pl=%d",
|
||||
kind, fmt_code(code), mid, tok.hex() or '-',
|
||||
len(ropts), len(payload))
|
||||
|
||||
# ACK back any CON from the device to suppress retransmits.
|
||||
# RFC 7252 §4.2 — ACK is a bare frame (token len 0, code 0).
|
||||
if mt == TYPE_CON:
|
||||
try:
|
||||
self._send_dgram(build_coap(TYPE_ACK, 0, mid, b'', []))
|
||||
except Exception as e:
|
||||
logger.warning("ACK send: %s", e)
|
||||
|
||||
# Empty ACK with no options & no payload = "separate response
|
||||
# coming" — used by Samsung's RT-OCF for the larger reads. Stop
|
||||
# the retransmit timer on the client side and wait for the CON.
|
||||
if mt == TYPE_ACK and code == 0 and not payload and not ropts:
|
||||
return
|
||||
|
||||
# Pending one-shot? Resolve and return.
|
||||
rec = self._pending.get(tok)
|
||||
if rec is not None:
|
||||
ev, container = rec
|
||||
container['code'] = code
|
||||
container['mtype'] = mt
|
||||
container['mid'] = mid
|
||||
container['options'] = ropts
|
||||
container['payload'] = payload
|
||||
ev.set()
|
||||
return
|
||||
|
||||
# OBSERVE notification?
|
||||
href = self._observe_tokens.get(tok)
|
||||
if href is not None:
|
||||
if code != 0x45:
|
||||
logger.warning("observe %s: non-2.05 %s",
|
||||
href, fmt_code(code))
|
||||
return
|
||||
cb = self.on_notification
|
||||
if cb is not None:
|
||||
try:
|
||||
cb(href, payload)
|
||||
except Exception as e:
|
||||
logger.warning("notification callback %s: %s",
|
||||
href, e)
|
||||
return
|
||||
|
||||
# Stale token (post-reconnect or unknown) — drop quietly.
|
||||
|
||||
# ---- request primitives ------------------------------------------
|
||||
|
||||
def get(self, path_segs, query=(), timeout=10.0):
|
||||
"""Token-stable Block2 GET. Returns (code, payload_bytes).
|
||||
|
||||
Reuses one CoAP token across every block of a multi-block
|
||||
response — Samsung's server keys per-transfer state on the
|
||||
token, and dropping a fresh token on block 1+ silently drops
|
||||
the request."""
|
||||
if self.conn is None:
|
||||
raise ConnectionError("DTLS session closed")
|
||||
tok = self._next_tok()
|
||||
blob = b''
|
||||
num = 0
|
||||
last_code = None
|
||||
last_opts = []
|
||||
deadline = time.time() + timeout
|
||||
szx = BLOCK_SZX # server may negotiate down; track per-transfer
|
||||
while True:
|
||||
if num > 0:
|
||||
self.pace()
|
||||
container = {}
|
||||
for attempt in range(_BLOCK_MAX_ATTEMPTS):
|
||||
ev = threading.Event()
|
||||
container = {}
|
||||
self._pending[tok] = (ev, container)
|
||||
try:
|
||||
mid = self._next_mid()
|
||||
opts = [(URI_PATH, s.encode()) for s in path_segs]
|
||||
for q in query:
|
||||
opts.append((URI_QUERY, q.encode()))
|
||||
opts.append((ACCEPT, CF_CBOR))
|
||||
if num > 0:
|
||||
opts.append((BLOCK2, block_value(num, 0, szx)))
|
||||
self._send_dgram(
|
||||
build_coap(TYPE_CON, METHOD_GET, mid, tok, opts))
|
||||
per_wait = min(_BLOCK_ACK_TIMEOUT,
|
||||
max(0.1, deadline - time.time()))
|
||||
if ev.wait(per_wait):
|
||||
break # got a response
|
||||
remaining = deadline - time.time()
|
||||
if remaining <= 0 or attempt == _BLOCK_MAX_ATTEMPTS - 1:
|
||||
logger.debug(
|
||||
"GET %s /%s block %d: timed out after %d attempt(s)",
|
||||
self.host, '/'.join(path_segs), num, attempt + 1,
|
||||
)
|
||||
raise TimeoutError(
|
||||
f"GET /{'/'.join(path_segs)} block {num} timeout")
|
||||
logger.debug(
|
||||
"GET %s /%s block %d: attempt %d/%d timeout, retrying",
|
||||
self.host, '/'.join(path_segs), num,
|
||||
attempt + 1, _BLOCK_MAX_ATTEMPTS,
|
||||
)
|
||||
finally:
|
||||
self._pending.pop(tok, None)
|
||||
if 'err' in container:
|
||||
raise ConnectionError(container['err'])
|
||||
|
||||
code = container['code']
|
||||
payload = container['payload']
|
||||
ropts = container['options']
|
||||
last_code = code
|
||||
last_opts = ropts
|
||||
blob += payload
|
||||
# 4.xx / 5.xx responses don't carry Block2 continuation —
|
||||
# bail with whatever we got. Caller decides if 4.xx is fatal.
|
||||
if code >> 5 != 2:
|
||||
return code, blob
|
||||
b2 = [v for n, v in ropts if n == BLOCK2]
|
||||
more = 0
|
||||
if b2:
|
||||
bv = int.from_bytes(b2[0], 'big')
|
||||
more = (bv >> 3) & 1
|
||||
server_szx = bv & 0x07
|
||||
if server_szx != szx:
|
||||
szx = server_szx
|
||||
if not more:
|
||||
break
|
||||
num += 1
|
||||
if num > self.MAX_BLOCKS:
|
||||
raise ConnectionError(
|
||||
f"GET /{'/'.join(path_segs)}: >{self.MAX_BLOCKS} "
|
||||
f"blocks, aborting")
|
||||
return last_code, blob
|
||||
|
||||
def post(self, path_segs, body_cbor, timeout=8.0):
|
||||
"""Single-frame POST with a CBOR-encoded body. Returns
|
||||
(code, payload_bytes). body_cbor must already be encoded."""
|
||||
if self.conn is None:
|
||||
raise ConnectionError("DTLS session closed")
|
||||
tok = self._next_tok()
|
||||
mid = self._next_mid()
|
||||
opts = [(URI_PATH, s.encode()) for s in path_segs]
|
||||
opts.append((CONTENT_FORMAT, CF_CBOR))
|
||||
opts.append((ACCEPT, CF_CBOR))
|
||||
datagram = build_coap(TYPE_CON, METHOD_POST, mid, tok, opts,
|
||||
body_cbor)
|
||||
ev = threading.Event()
|
||||
container = {}
|
||||
self._pending[tok] = (ev, container)
|
||||
try:
|
||||
self._send_dgram(datagram)
|
||||
if not ev.wait(timeout):
|
||||
raise TimeoutError(
|
||||
f"POST /{'/'.join(path_segs)} timeout")
|
||||
if 'err' in container:
|
||||
raise ConnectionError(container['err'])
|
||||
return container['code'], container['payload']
|
||||
finally:
|
||||
self._pending.pop(tok, None)
|
||||
|
||||
def ping(self):
|
||||
"""RFC 7252 §4.4 CoAP Ping — empty CON, no token, no payload.
|
||||
Fire-and-forget: we do not wait for the matching RST because
|
||||
Samsung's RT-OCF doesn't reliably emit one (verified
|
||||
2026-06-04: every sync ping timed out while polls succeeded
|
||||
at 200+/window). The send itself is the keepalive — it
|
||||
tickles Samsung's observer state so OBSERVE subscriptions
|
||||
aren't aged out.
|
||||
|
||||
Real half-open-session detection lives in PollScheduler's
|
||||
`last_success_ts`, surfaced through KeepaliveTask's
|
||||
`liveness_fn`."""
|
||||
if self.conn is None:
|
||||
raise ConnectionError("DTLS session closed")
|
||||
mid = self._next_mid()
|
||||
self._send_dgram(build_coap(TYPE_CON, 0, mid, b'', []))
|
||||
return mid
|
||||
|
||||
def refresh_observes(self, paths):
|
||||
"""Drop all current OBSERVE registrations and re-subscribe to
|
||||
the given paths. Used as a periodic safety net — CoAP OBSERVE
|
||||
has no built-in TTL but Samsung's RT-OCF can age out its
|
||||
observer table during cloud blips even while the DTLS session
|
||||
stays healthy. Without this, internet recovery on a still-
|
||||
reachable device leaves push permanently dead.
|
||||
|
||||
Best-effort: dereg failures are logged and we still bind fresh
|
||||
tokens via subscribe. Brief race window where a notify on the
|
||||
old token gets dropped as 'stale' — acceptable for a 6h-scale
|
||||
safety net."""
|
||||
if self.conn is None:
|
||||
raise ConnectionError("DTLS session closed")
|
||||
for tok, href in list(self._observe_tokens.items()):
|
||||
segs = [s for s in href.split('/') if s]
|
||||
try:
|
||||
self._send_observe_dereg(tok, segs)
|
||||
except Exception as e:
|
||||
logger.warning("refresh dereg %s: %s", href, e)
|
||||
self._observe_tokens.clear()
|
||||
time.sleep(0.1)
|
||||
for path in paths:
|
||||
try:
|
||||
self.subscribe(list(path))
|
||||
time.sleep(0.05)
|
||||
except Exception as e:
|
||||
logger.warning("refresh subscribe %s: %s", path, e)
|
||||
|
||||
def subscribe(self, path_segs):
|
||||
"""Register an OBSERVE on the given path. The initial 2.05
|
||||
notification and all subsequent state-change notifications
|
||||
will fire on_notification(href, payload_bytes).
|
||||
|
||||
Returns the token used (in case the caller wants to deregister
|
||||
later)."""
|
||||
if self.conn is None:
|
||||
raise ConnectionError("DTLS session closed")
|
||||
tok = self._next_observe_tok()
|
||||
href = '/' + '/'.join(path_segs)
|
||||
# Register the token BEFORE sending — otherwise the device
|
||||
# could respond between send() and the dict insert, and the
|
||||
# reader thread would drop the initial 2.05 as "stale".
|
||||
self._observe_tokens[tok] = href
|
||||
mid = self._next_mid()
|
||||
opts = [(URI_PATH, s.encode()) for s in path_segs]
|
||||
opts.append((OBSERVE, OBSERVE_REGISTER))
|
||||
opts.append((ACCEPT, CF_CBOR))
|
||||
self._send_dgram(
|
||||
build_coap(TYPE_CON, METHOD_GET, mid, tok, opts))
|
||||
return tok
|
||||
@@ -0,0 +1,15 @@
|
||||
-----BEGIN CERTIFICATE-----
|
||||
MIICXTCCAgGgAwIBAgIBATAMBggqhkjOPQQDAgUAMGsxKDAmBgNVBAMTH1NhbXN1
|
||||
bmcgRWxlY3Ryb25pY3MgT0NGIFJvb3QgQ0ExFDASBgNVBAsTC09DRiBSb290IENB
|
||||
MRwwGgYDVQQKExNTYW1zdW5nIEVsZWN0cm9uaWNzMQswCQYDVQQGEwJLUjAgFw0x
|
||||
NjExMjQwMjU1MTFaGA8yMDY5MTIzMTE0NTk1OVowazEoMCYGA1UEAxMfU2Ftc3Vu
|
||||
ZyBFbGVjdHJvbmljcyBPQ0YgUm9vdCBDQTEUMBIGA1UECxMLT0NGIFJvb3QgQ0Ex
|
||||
HDAaBgNVBAoTE1NhbXN1bmcgRWxlY3Ryb25pY3MxCzAJBgNVBAYTAktSMFkwEwYH
|
||||
KoZIzj0CAQYIKoZIzj0DAQcDQgAEYsf/Kx+sUFBQESbuytTDPwLPIe0X/8/B1L7b
|
||||
3abxE9w0gQZAfI8WYUkKfNfP7HXh1M5SCnOkfwWraltGOKTeX6OBkTCBjjAOBgNV
|
||||
HQ8BAf8EBAMCAcYwMgYDVR0fBCswKTAnoCWgI4YhaHR0cDovL3Byb2RjYS5zYW1z
|
||||
dW5naW90cy5jb20vY3JsMA8GA1UdEwEB/wQFMAMBAf8wNwYIKwYBBQUHAQEEKzAp
|
||||
MCcGCCsGAQUFBzABhhtodHRwOi8vb2NzcC5zYW1zdW5naW90cy5jb20wDAYIKoZI
|
||||
zj0EAwIFAANIADBFAiARY9aSE30q31q5v8B4sJczBqOp7AsD9o8ZIuNmH7IwSwIh
|
||||
ALfR56jcXoFis/nDx0tQ2xTI/f0b7F6tWPqj2vyKQeNR
|
||||
-----END CERTIFICATE-----
|
||||
Reference in New Issue
Block a user