Sentry

Implement Sentry scanning, notification state and worker cancellation through explicit policies and preferences.

openpilot/starpilot/sentry_mode/README.md

@@ -0,0 +1,49 @@
+# Sentry motion events
+
+This is the parked motion-event part of Sentry. Galaxy can list local event
+metadata at `#/cameras/events`. Camera capture, image galleries, notifications
+and remote access are not implemented here. Motion detection has
+not yet been qualified on a physical device or vehicle.
+
+The manager runs it only on device hardware when both
+`STARPILOT_SENTRY_DEVELOPMENT=1` and a valid enabled `SentryMotionPreferences`
+document are present. Ordinary startup is unchanged. The same manager-owned
+`sensord` supplies acceleration onroad and, under that explicit opt-in, while
+parked. Sentry never starts another sensor, camera or power service.
+
+The optional version-1 document contains `enabled`, `sensitivity` (0.005–1.0)
+and `warningTimeSeconds` (0.1–10.0). Missing, unreadable or malformed settings
+disable observation without replacing the saved bytes. There is no automatic
+adoption of historical Sentry settings. Galaxy exposes the saved choices at
+`#/cameras/sentry-settings`, through Cameras. Each edit requires a one-time
+confirmation and fresh parked evidence, then compares the complete saved source
+again before replacement. Invalid readable settings require an explicit reset
+to Off and defaults; unreadable settings remain untouched. Saving preferences
+does not bypass development opt-in. The shared settings page shows runtime
+configuration blocks, a live arming countdown and motion-owner evidence failures.
+Status expires after 1.5 seconds, is bound to the exact saved settings and boot,
+and never grants sensor, camera or power authority. Heartbeats use namespace-local
+shared memory on device hardware, not persistent event storage.
+
+Arming requires 90 seconds of continuously fresh parked, ignition-off, power
+and motion evidence. Observation runs at 10 Hz, preserving the magnitude-delta
+threshold and warning-count behavior. An alarm requires more than 25 qualifying
+deltas and at least 30 seconds within a 60-second motion window. A stale or
+repeated sample, sensor loss, owner-loop gap, system resume, settings change,
+ignition or shutdown request disarms the policy. Recovery requires rearming.
+Hardwared retains battery accounting and shutdown authority; Sentry neither
+extends parked runtime nor clears shutdown requests.
+
+Warning and alarm records contain only kind, generated event/session IDs and
+monotonic/system-clock timestamps. System time is not proof of an accurate wall
+clock. Storage uses a private namespace-specific directory, at most 512 bounded
+records, and atomic publication without overwriting an existing event. Full or
+unavailable storage stops recording new events; it does not delete old evidence.
+There are no image paths, location data or network requests. Authority and saved
+settings are checked again immediately before publication. A publication whose
+directory sync fails reports uncertain durability and is never blindly retried.
+
+Tests cover continuous arming and thresholds, corrupt preferences, serialized
+message timing, resume and power/ignition revocation, process selection, bounded
+storage and interrupted writes. They do not establish physical motion sensitivity,
+parked battery endurance, device wake behavior or complete Sentry parity.

openpilot/starpilot/sentry_mode/__init__.py

@@ -0,0 +1 @@
+"""Parked Sentry motion policy, guarded observation and local event metadata."""

openpilot/starpilot/sentry_mode/actions.py

@@ -0,0 +1,14 @@
+"""One validated, exact-source saved Sentry document replacement."""
+
+from collections.abc import Callable
+
+from openpilot.starpilot.saved_document import WriteResult, commit_exact
+from openpilot.starpilot.sentry_mode.preferences import KEY, MAX_BYTES, decode, encode
+
+
+def commit(params, raw: bytes, expected: bytes | None, authorized: Callable[[], bool]) -> WriteResult:
+  preferences = decode(raw) if type(raw) is bytes else None
+  if preferences is None or encode(preferences) != raw:
+    return WriteResult(False, False)
+  return commit_exact(params, key=KEY, max_bytes=MAX_BYTES, raw=raw, expected=expected,
+                      authorized=authorized, temp_prefix=".tmp_sentry_")

openpilot/starpilot/sentry_mode/notifications.py

@@ -0,0 +1,451 @@
+"""Private configuration and durable, bounded event delivery outside motion detection.
+
+A successful receiver response is deduplicated durably. An interrupted request can
+have been accepted remotely: retries reuse the event/channel Idempotency-Key, but
+arbitrary webhooks and ntfy cannot promise exactly-once display.
+"""
+import copy
+import fcntl
+import hashlib
+import json
+import os
+from pathlib import Path
+import re
+import stat
+import threading
+import time
+from urllib.parse import urlsplit
+import uuid
+
+import requests
+
+from openpilot.starpilot.storage import starpilot_storage_root
+from openpilot.starpilot.sentry_mode.storage import EventStore
+from openpilot.starpilot.sentry_mode import web_push
+
+CHANNELS = ('webPush', 'discord', 'webhook', 'ntfy')
+MAX_STATE_BYTES = 1024 * 1024
+MAX_JOBS = 512
+MAX_AGE = 86400
+MAX_ATTEMPTS = 4
+RETRY_SECONDS = (30, 120, 600, 600)
+ID = re.compile(r'[0-9a-f]{32}\Z')
+
+
+class NotificationUnavailable(Exception):
+  pass
+
+
+def endpoint(value: object, *, allow_http: bool = False) -> str:
+  if type(value) is not str or not 1 <= len(value) <= 2048 or any(ord(c) <= 32 for c in value):
+    raise ValueError('Enter a valid notification URL')
+  try:
+    parsed = urlsplit(value)
+    if (parsed.scheme not in (('https', 'http') if allow_http else ('https',)) or not parsed.hostname or
+        parsed.username is not None or parsed.password is not None or parsed.fragment or parsed.port == 0):
+      raise ValueError
+  except ValueError:
+    raise ValueError('Use an HTTPS notification URL') from None
+  return value
+
+
+def fresh_state() -> dict:
+  return {'version': 1, 'channels': {channel: {'enabled': False, 'url': '', 'token': '', 'generation': uuid.uuid4().hex,
+                                              'since': 0} for channel in CHANNELS},
+          'subscriptions': [], 'vapid': '', 'jobs': {}, 'queueFull': False}
+
+
+def wall_seconds() -> float:
+  # Durable retry deadlines and event timestamps must survive a process/device reboot.
+  return time.time_ns() / 1e9
+
+
+class NotificationOwner:
+  def __init__(self, root: Path | None = None, store=None, *, clock=wall_seconds, transport=None, allow_http: bool = False):
+    self.root = Path(root) if root is not None else starpilot_storage_root() / "sentry/notifications"
+    self.store = store if store is not None else EventStore()
+    self.clock, self.allow_http = clock, allow_http
+    self.transport = transport if transport is not None else self._send
+    self.lock = threading.RLock()
+    self.stop = threading.Event()
+    self.thread = None
+    self.state = fresh_state()
+    self.available = True
+    self.lock_fd = None
+    self.saved_bytes = None
+    try:
+      directory = self._directory()
+      try:
+        self.lock_fd = os.open('.lock', os.O_CREAT | os.O_RDWR | os.O_NOFOLLOW | os.O_NONBLOCK, 0o600, dir_fd=directory)
+        lock_info = os.fstat(self.lock_fd)
+        if not stat.S_ISREG(lock_info.st_mode) or lock_info.st_uid != os.getuid() or stat.S_IMODE(lock_info.st_mode) & 0o077:
+          raise OSError('Invalid notification lock')
+        fcntl.flock(self.lock_fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
+      finally:
+        os.close(directory)
+      self._load()
+    except (OSError, ValueError, TypeError, RecursionError):
+      self.available = False
+      if self.lock_fd is not None:
+        os.close(self.lock_fd)
+        self.lock_fd = None
+
+  def _directory(self) -> int:
+    # Walk every component without following symlinks. This private state contains credentials.
+    if not self.root.is_absolute() or ".." in self.root.parts:
+      raise OSError('Private storage requires an absolute directory')
+    fd = os.open('/', os.O_RDONLY | os.O_DIRECTORY)
+    try:
+      for component in self.root.parts[1:]:
+        try:
+          os.mkdir(component, 0o700, dir_fd=fd)
+        except FileExistsError:
+          pass
+        child = os.open(component, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW, dir_fd=fd)
+        os.close(fd)
+        fd = child
+      info = os.fstat(fd)
+      if info.st_uid != os.getuid() or stat.S_IMODE(info.st_mode) & 0o077:
+        raise OSError('Private storage permissions are invalid')
+      return fd
+    except BaseException:
+      os.close(fd)
+      raise
+
+  def _load(self) -> None:
+    directory = self._directory()
+    try:
+      try:
+        fd = os.open('state.json', os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK, dir_fd=directory)
+      except FileNotFoundError:
+        return
+      try:
+        info = os.fstat(fd)
+        if not stat.S_ISREG(info.st_mode) or info.st_nlink != 1 or info.st_uid != os.getuid() or stat.S_IMODE(info.st_mode) & 0o077:
+          raise ValueError('Invalid private notification state')
+        raw = os.read(fd, MAX_STATE_BYTES + 1)
+      finally:
+        os.close(fd)
+      if len(raw) > MAX_STATE_BYTES:
+        raise ValueError('Notification state exceeds limit')
+      def unique(pairs):
+        result = {}
+        for key, value in pairs:
+          if key in result:
+            raise ValueError('Duplicate notification state')
+          result[key] = value
+        return result
+      state = json.loads(raw, object_pairs_hook=unique)
+      self._validate_state(state)
+      self.state = state
+      self.saved_bytes = json.dumps(state, separators=(',', ':'), allow_nan=False).encode()
+    finally:
+      os.close(directory)
+
+  def _validate_state(self, state: dict) -> None:
+    if (type(state) is not dict or set(state) != set(fresh_state()) or type(state['version']) is not int or state['version'] != 1 or
+        type(state['channels']) is not dict or set(state['channels']) != set(CHANNELS) or
+        type(state['subscriptions']) is not list or len(state['subscriptions']) > 16 or
+        type(state['jobs']) is not dict or len(state['jobs']) > MAX_JOBS or type(state['queueFull']) is not bool):
+      raise ValueError('Invalid notification state')
+    if type(state['vapid']) is not str or len(state['vapid']) > 4096:
+      raise ValueError('Invalid notification push key')
+    if state['vapid']:
+      web_push.load_key(state['vapid'])
+    for name, channel in state['channels'].items():
+      if (type(channel) is not dict or set(channel) != {'enabled', 'url', 'token', 'generation', 'since'} or
+          type(channel['enabled']) is not bool or type(channel['generation']) is not str or ID.fullmatch(channel['generation']) is None or
+          type(channel['since']) not in (int, float) or not 0 <= channel['since'] < 2**63 or
+          type(channel['url']) is not str or len(channel['url']) > 2048 or
+          type(channel['token']) is not str or len(channel['token']) > 512 or any(ord(c) < 32 for c in channel['token'])):
+        raise ValueError('Invalid notification channel')
+      if channel['url']:
+        endpoint(channel['url'], allow_http=self.allow_http)
+      if name != 'webPush' and channel['enabled'] and not channel['url']:
+        raise ValueError('Missing notification endpoint')
+    for sub in state['subscriptions']:
+      if type(sub) is not dict or set(sub) != {'id', 'subscription', 'since'} or type(sub['id']) is not str or ID.fullmatch(sub['id']) is None:
+        raise ValueError('Invalid browser subscription')
+      web_push.validate_subscription(sub['subscription'])
+      endpoint(sub['subscription']['endpoint'], allow_http=self.allow_http)
+      if type(sub['since']) not in (int, float) or not 0 <= sub['since'] < 2**63:
+        raise ValueError('Invalid subscription time')
+    for key, job in state['jobs'].items():
+      if (type(job) is not dict or set(job) != {'eventId', 'kind', 'channel', 'generation', 'subscriptionId', 'created',
+                                               'next', 'attempts', 'state', 'error'} or
+          type(key) is not str or len(key) != 64 or type(job['eventId']) is not str or ID.fullmatch(job['eventId']) is None or
+          job['kind'] not in ('warning', 'alarm', 'test') or job['channel'] not in CHANNELS or
+          type(job['generation']) is not str or ID.fullmatch(job['generation']) is None or
+          (type(job['subscriptionId']) is not str or (job['subscriptionId'] and ID.fullmatch(job['subscriptionId']) is None)) or
+          type(job['attempts']) is not int or not 0 <= job['attempts'] <= MAX_ATTEMPTS or
+          job['state'] not in ('queued', 'sending', 'sent', 'failed', 'cancelled') or
+          job['error'] not in ('', 'http', 'expired', 'timeout', 'transport', 'delivery_unknown', 'stale') or
+          any(type(job[k]) not in (float, int) or not 0 <= job[k] < 2**63 for k in ('created', 'next'))):
+        raise ValueError('Invalid notification job')
+
+  def _save(self) -> None:
+    raw = json.dumps(self.state, separators=(',', ':'), allow_nan=False).encode()
+    if len(raw) > MAX_STATE_BYTES:
+      raise NotificationUnavailable('Notification storage is full')
+    if raw == self.saved_bytes:
+      return
+    directory = self._directory()
+    temporary = '.' + uuid.uuid4().hex + '.tmp'
+    try:
+      fd = os.open(temporary, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600, dir_fd=directory)
+      with os.fdopen(fd, 'wb') as output:
+        output.write(raw)
+        output.flush()
+        os.fsync(output.fileno())
+      os.replace(temporary, 'state.json', src_dir_fd=directory, dst_dir_fd=directory)
+      os.fsync(directory)
+      self.saved_bytes = raw
+    except OSError:
+      self.available = False
+      raise NotificationUnavailable('Notification storage is unavailable') from None
+    finally:
+      try:
+        os.unlink(temporary, dir_fd=directory)
+      except FileNotFoundError:
+        pass
+      os.close(directory)
+
+  def snapshot(self) -> dict:
+    with self.lock:
+      if not self.available:
+        raise NotificationUnavailable('Notification storage is unavailable')
+      jobs = self.state['jobs']
+      result = {'schemaVersion': 1, 'channels': {}, 'queueFull': self.state['queueFull'],
+                'subscriptionCount': len(self.state['subscriptions']),
+                'subscriptions': [{'id': sub['id']} for sub in self.state['subscriptions']],
+                'deliverySemantics': 'Retries may repeat delivery after an interrupted remote response.'}
+      for name, cfg in self.state['channels'].items():
+        relevant = [j for j in jobs.values() if j['channel'] == name and j['generation'] == cfg['generation']]
+        latest = max(relevant, key=lambda j: j['created'], default=None)
+        result['channels'][name] = {'enabled': cfg['enabled'],
+                                    'configured': bool(self.state['subscriptions']) if name == 'webPush' else bool(cfg['url']),
+                                    'pending': sum(j['state'] in ('queued', 'sending') for j in relevant),
+                                    'lastState': latest['state'] if latest else 'idle',
+                                    'lastError': latest['error'] if latest else ''}
+      if self.state['vapid']:
+        result['applicationServerKey'] = web_push.b64(web_push.public(web_push.load_key(self.state['vapid'])))
+      return result
+
+  def action(self, payload: dict, *, permitted=lambda: True) -> dict:
+    with self.lock:
+      if not self.available:
+        raise NotificationUnavailable('Notification storage is unavailable')
+      if not permitted():
+        raise PermissionError('A current Galaxy session is required')
+      if type(payload) is not dict or type(payload.get('action')) is not str:
+        raise ValueError('Choose a notification action')
+      previous = copy.deepcopy(self.state)
+      action = payload['action']
+      now = self.clock()
+      if action == 'configure':
+        if set(payload) != {'action', 'channel', 'enabled', 'url', 'token'} or payload['channel'] not in CHANNELS:
+          raise ValueError('Invalid notification channel')
+        name = payload['channel']
+        if (type(payload['enabled']) is not bool or type(payload['token']) is not str or len(payload['token']) > 512 or
+            any(ord(c) < 32 for c in payload['token'])):
+          raise ValueError('Invalid notification setting')
+        url = payload['url']
+        if type(url) is not str:
+          raise ValueError('Enter a valid notification URL')
+        if url:
+          endpoint(url, allow_http=self.allow_http)
+        if name == 'webPush' and payload['enabled'] and not self.state['subscriptions']:
+          raise ValueError('Subscribe a browser first')
+        if name == 'webPush' and (url or payload['token']):
+          raise ValueError('Browser push uses a browser subscription')
+        old = self.state['channels'][name]
+        # Empty fields preserve configured credentials; explicit disable does not delete them.
+        url = url or old['url']
+        token = payload['token'] or old['token']
+        if payload['enabled'] and name != 'webPush' and not url:
+          raise ValueError('Enter a notification URL')
+        self.state['channels'][name] = {'enabled': payload['enabled'], 'url': url, 'token': token,
+                                        'generation': uuid.uuid4().hex, 'since': now}
+      elif action == 'forget':
+        if set(payload) != {'action', 'channel'} or payload['channel'] not in CHANNELS:
+          raise ValueError('Choose a notification channel')
+        self.state['channels'][payload['channel']] = fresh_state()['channels'][payload['channel']]
+        if payload['channel'] == 'webPush':
+          self.state['subscriptions'] = []
+      elif action == 'pushKey':
+        if set(payload) != {'action'}:
+          raise ValueError('Invalid push key request')
+        if not self.state['vapid']:
+          self.state['vapid'] = web_push.new_key()
+      elif action == 'subscribe':
+        if set(payload) != {'action', 'subscription'} or not self.state['vapid']:
+          raise ValueError('Prepare browser push first')
+        sub = payload['subscription']
+        web_push.validate_subscription(sub)
+        endpoint(sub['endpoint'], allow_http=self.allow_http)
+        current = self.state['subscriptions']
+        existing = next((s for s in current if s['subscription']['endpoint'] == sub['endpoint']), None)
+        if existing:
+          existing['subscription'] = sub
+        elif len(current) < 16:
+          current.append({'id': uuid.uuid4().hex, 'subscription': sub, 'since': now})
+        else:
+          raise ValueError('Remove a browser before adding another')
+      elif action == 'unsubscribe':
+        if set(payload) != {'action', 'id'} or type(payload['id']) is not str or ID.fullmatch(payload['id']) is None:
+          raise ValueError('Choose a browser subscription')
+        self.state['subscriptions'] = [s for s in self.state['subscriptions'] if s['id'] != payload['id']]
+      elif action == 'test':
+        if set(payload) != {'action', 'channel'} or payload['channel'] not in CHANNELS:
+          raise ValueError('Choose a test channel')
+        cfg = self.state['channels'][payload['channel']]
+        if not cfg['enabled']:
+          raise ValueError('Enable this notification channel first')
+        if payload['channel'] == 'webPush' and not self.state['subscriptions']:
+          raise ValueError('Subscribe a browser first')
+        if sum(j['kind'] == 'test' and j['created'] > now - 60 for j in self.state['jobs'].values()) >= 4:
+          raise ValueError('Wait a minute before sending more tests')
+        self._enqueue({'eventId': uuid.uuid4().hex, 'kind': 'test', 'wallTimeNs': int(now * 1e9)}, only=payload['channel'])
+      else:
+        raise ValueError('Choose a supported notification action')
+      if not permitted():
+        self.state = previous
+        raise PermissionError('Galaxy session changed')
+      self._save()
+      return self.snapshot()
+
+  def _enqueue(self, event: dict, *, only: str | None = None) -> None:
+    now = self.clock()
+    for name, cfg in self.state['channels'].items():
+      if not cfg['enabled'] or (only is not None and only != name):
+        continue
+      targets = self.state['subscriptions'] if name == 'webPush' else [{'id': '', 'since': cfg['since']}]
+      for target in targets:
+        if event['kind'] != 'test' and event['wallTimeNs'] / 1e9 <= max(cfg['since'], target['since']):
+          continue
+        key = hashlib.sha256((event['eventId'] + name + cfg['generation'] + target['id']).encode()).hexdigest()
+        if key in self.state['jobs']:
+          continue
+        if len(self.state['jobs']) >= MAX_JOBS:
+          self.state['queueFull'] = True
+          continue
+        self.state['jobs'][key] = {'eventId': event['eventId'], 'kind': event['kind'], 'channel': name,
+                                  'generation': cfg['generation'], 'subscriptionId': target['id'],
+                                  'created': event['wallTimeNs'] / 1e9, 'next': now, 'attempts': 0,
+                                  'state': 'queued', 'error': ''}
+
+  def tick(self) -> None:
+    with self.lock:
+      if not self.available:
+        return
+      now = self.clock()
+      self.state['jobs'] = {key: job for key, job in self.state['jobs'].items() if now - job['created'] <= MAX_AGE}
+      self.state['queueFull'] = False
+      if not any(cfg["enabled"] for cfg in self.state["channels"].values()):
+        return
+      snapshot = self.store.snapshot()
+      for event in snapshot['events']:
+        if 0 <= now - event['wallTimeNs'] / 1e9 <= MAX_AGE:
+          self._enqueue(event)
+      for key, job in self.state['jobs'].items():
+        if job['state'] not in ('queued', 'sending'):
+          continue
+        cfg = self.state['channels'][job['channel']]
+        sub = next((s['subscription'] for s in self.state['subscriptions'] if s['id'] == job['subscriptionId']), None)
+        if not cfg['enabled'] or cfg['generation'] != job['generation'] or (job['channel'] == 'webPush' and sub is None):
+          job['state'] = 'cancelled'
+          continue
+        if now < job['created'] or now - job['created'] > MAX_AGE:
+          job['state'], job['error'] = 'failed', 'stale'
+          continue
+        if job['state'] == 'sending':
+          job['error'] = 'delivery_unknown'
+          job['state'] = 'queued' if job['attempts'] < MAX_ATTEMPTS else 'failed'
+          job['next'] = now + RETRY_SECONDS[min(job['attempts'], MAX_ATTEMPTS) - 1]
+        if job['state'] != 'queued' or job['next'] > now:
+          continue
+        job['attempts'] += 1
+        job['state'] = 'sending'
+        self._save()  # Claim durably before issuing external IO; secrets stay in private config.
+        try:
+          # Keep status/configuration responsive while network IO is in flight.
+          self.lock.release()
+          try:
+            code = self.transport(job['channel'], dict(cfg), sub, dict(job), key, self.state['vapid'])
+          finally:
+            self.lock.acquire()
+            job = self.state['jobs'].get(key, job)
+          if 200 <= code < 300:
+            job['state'], job['error'] = 'sent', ''
+          elif code in (404, 410) and job['channel'] == 'webPush':
+            job['state'], job['error'] = 'failed', 'expired'
+            self.state['subscriptions'] = [s for s in self.state['subscriptions']
+                                           if s['id'] != job['subscriptionId'] or s['subscription'] != sub]
+          else:
+            job['state'], job['error'] = ('queued' if code in (408, 429) or code >= 500 else 'failed'), 'http'
+        except (requests.Timeout, TimeoutError):
+          job['state'], job['error'] = 'queued', 'timeout'
+        except Exception:
+          # Never log response bodies, URLs, exception text, subscription secrets or auth headers.
+          job['state'], job['error'] = 'queued', 'transport'
+        if job['state'] == 'queued':
+          if job['attempts'] >= MAX_ATTEMPTS:
+            job['state'] = 'failed'
+          job['next'] = now + RETRY_SECONDS[job['attempts'] - 1]
+        self._save()
+        return  # At most one HTTP attempt per tick, per owner.
+      self._save()
+
+  def _send(self, name: str, cfg: dict, subscription: dict | None, job: dict, key: str, pem: str) -> int:
+    message = 'This is a test StarPilot Sentry notification.' if job['kind'] == 'test' else f"Parked motion detected ({job['kind']})."
+    event = {'eventId': job['eventId'], 'kind': job['kind'], 'systemTimeMs': int(job['created'] * 1000)}
+    headers = {'Idempotency-Key': key}
+    kwargs = {'timeout': (3, 10), 'allow_redirects': False, 'stream': True}
+    if name == 'webPush':
+      body, push_headers = web_push.request(subscription, {'title': 'StarPilot Sentry Mode', 'body': message,
+                                            'eventId': job['eventId'], 'url': '/#/cameras/events'}, pem)
+      headers.update(push_headers)
+      response = requests.post(subscription['endpoint'], data=body, headers=headers, **kwargs)
+    else:
+      if cfg['token']:
+        headers['Authorization'] = 'Bearer ' + cfg['token']
+      if name == 'discord':
+        response = requests.post(cfg['url'], json={'content': 'StarPilot Sentry Mode: ' + message}, headers=headers, **kwargs)
+      elif name == 'webhook':
+        # Preserve the original general-webhook content/event form contract.
+        response = requests.post(cfg['url'], data={'content': 'StarPilot Sentry Mode: ' + message,
+                                  'event': json.dumps(event, separators=(',', ':'))}, headers=headers, **kwargs)
+      else:
+        headers.update({'Title': 'StarPilot Sentry Mode', 'Priority': 'urgent', 'Tags': 'warning,car'})
+        response = requests.post(cfg['url'], data=message.encode(), headers=headers, **kwargs)
+    try:
+      return response.status_code
+    finally:
+      response.close()
+
+  def start(self) -> None:
+    with self.lock:
+      if self.thread is not None:
+        return
+      def run():
+        while not self.stop.is_set():
+          try:
+            self.tick()
+          except Exception:
+            # Storage/scanner failure is visible, never rewritten as a successful delivery.
+            with self.lock:
+              self.available = False
+            return
+          self.stop.wait(2.0)
+      self.thread = threading.Thread(target=run, name='sentry-notifications', daemon=True)
+      self.thread.start()
+
+  def close(self) -> None:
+    self.stop.set()
+    if self.thread is not None:
+      self.thread.join(timeout=15)
+      if self.thread.is_alive():
+        return
+    if self.lock_fd is not None:
+      os.close(self.lock_fd)
+      self.lock_fd = None

openpilot/starpilot/sentry_mode/policy.py

@@ -0,0 +1,163 @@
+"""Parked Sentry arming and motion escalation without hardware or I/O."""
+
+from dataclasses import dataclass
+import math
+
+
+ARM_DELAY_SECONDS = 90.0
+LOOP_INTERVAL_SECONDS = 0.1
+ALARM_TRIGGER_COUNT = 25
+ALARM_TIME_SECONDS = 30.0
+RESET_TIME_SECONDS = 60.0
+MAX_SAMPLE_AGE_SECONDS = 0.5
+MAX_UPDATE_GAP_SECONDS = 0.5  # 10 Hz owner may drop a few ticks, not a sleep/resume interval.
+MAX_ACCELERATION_AXIS = 200.0  # Larger than plausible vehicle acceleration; rejects overflow and corrupt samples.
+
+
+@dataclass(frozen=True)
+class Settings:
+  sensitivity: float = 0.04
+  warning_time_seconds: float = 1.0
+
+  def __post_init__(self) -> None:
+    if type(self.sensitivity) not in (int, float) or not math.isfinite(self.sensitivity) or \
+       not 0.005 <= self.sensitivity <= 1.0:
+      raise ValueError("Sentry sensitivity must be within 0.005–1.0")
+    if type(self.warning_time_seconds) not in (int, float) or not math.isfinite(self.warning_time_seconds) or \
+       not 0.1 <= self.warning_time_seconds <= 10.0:
+      raise ValueError("Sentry warning time must be within 0.1–10 seconds")
+
+  @property
+  def warning_trigger_count(self) -> int:
+    # Count 10 Hz samples; this threshold is not elapsed time.
+    return max(1, math.ceil(self.warning_time_seconds / LOOP_INTERVAL_SECONDS))
+
+
+@dataclass(frozen=True)
+class MotionSample:
+  mono_time: float
+  acceleration: tuple[float, float, float]
+
+  def __post_init__(self) -> None:
+    if type(self.mono_time) not in (int, float) or not math.isfinite(self.mono_time) or self.mono_time < 0 or \
+       not isinstance(self.acceleration, tuple) or len(self.acceleration) != 3 or \
+       any(type(axis) not in (int, float) or not math.isfinite(axis) or abs(axis) > MAX_ACCELERATION_AXIS
+           for axis in self.acceleration):
+      raise ValueError("Motion sample needs a finite monotonic timestamp and three finite axes")
+
+
+@dataclass(frozen=True)
+class Inputs:
+  enabled: bool
+  offroad: bool
+  voltage_ok: bool
+  device_fresh: bool
+  ignition_off: bool
+  sample: MotionSample | None = None
+
+  def __post_init__(self) -> None:
+    if any(type(value) is not bool for value in (self.enabled, self.offroad, self.voltage_ok,
+                                                self.device_fresh, self.ignition_off)) or \
+       (self.sample is not None and not isinstance(self.sample, MotionSample)):
+      raise ValueError("Sentry authority inputs must be explicit booleans and a validated motion sample")
+
+
+@dataclass(frozen=True)
+class Decision:
+  state: str
+  event: str | None = None
+  seconds_remaining: int | None = None
+
+
+class SentryPolicy:
+  def __init__(self, settings: Settings | None = None):
+    self.settings = settings if settings is not None else Settings()
+    self.last_now: float | None = None
+    self.arm_started: float | None = None
+    self.last_sample_time: float | None = None
+    self.previous_magnitude: float | None = None
+    self.trigger_count = 0
+    self.trigger_started: float | None = None
+    self.alarm_triggered = False
+
+  def _clear_motion(self) -> None:
+    self.last_sample_time = None
+    self.previous_magnitude = None
+    self.trigger_count = 0
+    self.trigger_started = None
+    self.alarm_triggered = False
+
+  def _disarm(self) -> None:
+    self.arm_started = None
+    self._clear_motion()
+
+  def update(self, now: float, inputs: Inputs) -> Decision:
+    if type(now) not in (int, float) or not math.isfinite(now) or now < 0:
+      self._disarm()
+      self.last_now = None
+      return Decision("unavailable")
+    if self.last_now is not None and now < self.last_now:
+      self._disarm()
+      self.last_now = now
+      return Decision("unavailable")
+    if self.last_now is not None and now - self.last_now > MAX_UPDATE_GAP_SECONDS:
+      self._disarm()
+      self.last_now = now
+      return Decision("unavailable")
+    self.last_now = now
+
+    if not inputs.enabled:
+      self._disarm()
+      return Decision("disabled")
+    if not inputs.offroad:
+      self._disarm()
+      return Decision("disabled_onroad")
+    if not inputs.ignition_off:
+      self._disarm()
+      return Decision("disabled_ignition")
+    if not inputs.voltage_ok:
+      self._disarm()
+      return Decision("low_voltage")
+    if not inputs.device_fresh:
+      self._disarm()
+      return Decision("unavailable")
+
+    sample = inputs.sample
+    if sample is None or sample.mono_time > now or now - sample.mono_time > MAX_SAMPLE_AGE_SECONDS or \
+       (self.last_sample_time is not None and sample.mono_time <= self.last_sample_time):
+      last_sample_time = self.last_sample_time
+      self._disarm()
+      # A duplicate may still be younger than the age limit on the next tick.
+      self.last_sample_time = last_sample_time
+      return Decision("sensor_unavailable")
+    self.last_sample_time = sample.mono_time
+
+    if self.arm_started is None:
+      self.arm_started = now
+    elapsed = now - self.arm_started
+    if elapsed < ARM_DELAY_SECONDS:
+      return Decision("arming", seconds_remaining=max(0, int(ARM_DELAY_SECONDS - elapsed)))
+
+    magnitude = math.hypot(*sample.acceleration)
+    previous = self.previous_magnitude
+    self.previous_magnitude = magnitude
+    if previous is None:
+      return Decision("armed")
+
+    # An expired motion window cannot turn into a late alarm on this sample.
+    if self.trigger_started is not None and now - self.trigger_started >= RESET_TIME_SECONDS:
+      self.trigger_count = 0
+      self.trigger_started = None
+      self.alarm_triggered = False
+
+    if abs(magnitude - previous) <= self.settings.sensitivity:
+      return Decision("armed")
+    self.trigger_count += 1
+    if self.trigger_started is None:
+      self.trigger_started = now
+    if self.trigger_count == self.settings.warning_trigger_count:
+      return Decision("armed", event="warning")
+    if self.trigger_count > ALARM_TRIGGER_COUNT and now - self.trigger_started >= ALARM_TIME_SECONDS and not self.alarm_triggered:
+      self.alarm_triggered = True
+      return Decision("armed", event="alarm")
+    return Decision("armed")

openpilot/starpilot/sentry_mode/preferences.py

@@ -0,0 +1,82 @@
+"""Strict optional saved Sentry motion policy; no legacy-key adoption."""
+
+from dataclasses import dataclass, field
+import json
+import math
+from typing import Any
+
+from openpilot.starpilot.saved_source import read_saved
+from openpilot.starpilot.sentry_mode.policy import Settings
+
+
+KEY = "SentryMotionPreferences"
+MAX_BYTES = 512
+
+
+@dataclass(frozen=True)
+class Preferences:
+  enabled: bool = False
+  settings: Settings = field(default_factory=Settings)
+
+
+@dataclass(frozen=True)
+class SavedPreferences:
+  raw: bytes | None
+  readable: bool
+  valid: bool
+  preferences: Preferences = field(default_factory=Preferences)
+
+
+def _unique(pairs: list[tuple[str, object]]) -> dict[str, object]:
+  result: dict[str, object] = {}
+  for key, value in pairs:
+    if key in result:
+      raise ValueError("duplicate Sentry preference")
+    result[key] = value
+  return result
+
+
+def decode(raw: bytes) -> Preferences | None:
+  if len(raw) > MAX_BYTES:
+    return None
+  try:
+    data = json.loads(raw, object_pairs_hook=_unique,
+                      parse_constant=lambda _: (_ for _ in ()).throw(ValueError("nonfinite")))
+    if not isinstance(data, dict) or set(data) != {"version", "enabled", "sensitivity", "warningTimeSeconds"}:
+      return None
+    if type(data["version"]) is not int or data["version"] != 1 or type(data["enabled"]) is not bool:
+      return None
+    sensitivity, warning = data["sensitivity"], data["warningTimeSeconds"]
+    if (type(sensitivity) not in (float, int) or type(warning) not in (float, int) or
+        not math.isfinite(sensitivity) or not math.isfinite(warning)):
+      return None
+    return Preferences(data["enabled"], Settings(float(sensitivity), float(warning)))
+  except (UnicodeDecodeError, ValueError, TypeError, OverflowError, RecursionError):
+    return None
+
+
+def encode(preferences: Preferences) -> bytes:
+  if not isinstance(preferences, Preferences) or type(preferences.enabled) is not bool:
+    raise ValueError("invalid Sentry preferences")
+  settings = Settings(preferences.settings.sensitivity, preferences.settings.warning_time_seconds)
+  data = {"version": 1, "enabled": preferences.enabled,
+          "sensitivity": settings.sensitivity, "warningTimeSeconds": settings.warning_time_seconds}
+  raw = json.dumps(data, sort_keys=True, separators=(",", ":"), allow_nan=False).encode()
+  if len(raw) > MAX_BYTES:
+    raise ValueError("oversized Sentry preferences")
+  return raw
+
+
+def read_preferences(params: Any) -> SavedPreferences:
+  raw, readable = read_saved(params, KEY, MAX_BYTES)
+  if not readable:
+    return SavedPreferences(raw, False, False)
+  if raw is None:
+    return SavedPreferences(None, True, True)
+  preferences = decode(raw)
+  return SavedPreferences(raw, True, preferences is not None, preferences or Preferences())
+
+
+def enabled(params: Any) -> bool:
+  saved = read_preferences(params)
+  return saved.readable and saved.valid and saved.preferences.enabled

openpilot/starpilot/sentry_mode/runtime.py

@@ -0,0 +1,215 @@
+"""Default-off parked motion observation; hardwared and sensord retain ownership."""
+
+from collections.abc import Callable
+import math
+import os
+import signal
+import threading
+import time
+from typing import Any
+
+from openpilot.common.params import Params
+from openpilot.common.swaglog import cloudlog
+from openpilot.starpilot.parked_evidence import ParkedEvidence, RESUME_SKEW_NS, fresh_parked
+from openpilot.starpilot.power.offroad_preferences import read_saved as read_power, effective as effective_power
+from openpilot.starpilot.saved_source import read_saved
+from openpilot.starpilot.sentry_mode.policy import Inputs, MotionSample, SentryPolicy
+from openpilot.starpilot.sentry_mode.preferences import read_preferences
+from openpilot.starpilot.sentry_mode.status import RuntimeStatus
+
+
+PERIPHERAL_TTL_NS = 1_500_000_000  # C++ 2 Hz publisher; one missed frame.
+SAMPLE_TTL_NS = 500_000_000
+TICK_SECONDS = 0.1
+
+
+def boot_time_ns() -> int:
+  return time.clock_gettime_ns(getattr(time, "CLOCK_BOOTTIME", time.CLOCK_MONOTONIC))
+
+
+class SentryRuntime:
+  def __init__(self, params: Params, messages: Any, *, mono_clock: Callable[[], int] = time.monotonic_ns,
+               boot_clock: Callable[[], int] = boot_time_ns,
+               record: Callable[[str, int, Callable[[], bool]], Any] | None = None, status: RuntimeStatus | None = None):
+    self.params, self.messages = params, messages
+    self.mono_clock, self.boot_clock = mono_clock, boot_clock
+    self.record = record
+    self.status = status
+    self.policy = SentryPolicy()
+    self.saved_raw: bytes | None = None
+    self.device_after_mono_ns = mono_clock()
+    self.sensor_after_mono_ns = self.device_after_mono_ns
+    self.device_offset_ns: int | None = None
+    self.last_pair_offset_ns: int | None = None
+    self.last_sample_ns = 0
+    self.recording_state = "idle"
+
+  def _recording_state(self, state: str) -> None:
+    if state != self.recording_state:
+      self.recording_state = state
+      if state not in ("idle", "stored"):
+        cloudlog.warning(f"sentry motion recording: {state}")
+
+  def _clocks(self) -> tuple[int, int] | None:
+    before = self.mono_clock()
+    boot = self.boot_clock()
+    after = self.mono_clock()
+    if after < before or after - before > RESUME_SKEW_NS:
+      self.device_after_mono_ns = max(self.device_after_mono_ns, after)
+      self.sensor_after_mono_ns = max(self.sensor_after_mono_ns, after)
+      self.device_offset_ns = None
+      self.policy = SentryPolicy(self.policy.settings)
+      return None
+    offset = boot - (before + after) // 2
+    if self.last_pair_offset_ns is not None and abs(offset - self.last_pair_offset_ns) > RESUME_SKEW_NS:
+      self.device_after_mono_ns = max(self.device_after_mono_ns, after)
+      self.sensor_after_mono_ns = max(self.sensor_after_mono_ns, after)
+      self.device_offset_ns = None
+      self.policy = SentryPolicy(self.policy.settings)
+    self.last_pair_offset_ns = offset
+    return after, boot
+
+  def _fresh(self, name: str, *, now_mono_ns: int, now_boot_ns: int, ttl_ns: int) -> bool:
+    sm = self.messages
+    stamp = int(sm.logMonoTime[name])
+    receipt = int(sm.recv_time[name] * 1e9)
+    if not (sm.seen[name] and sm.alive[name] and sm.valid[name] and stamp > 0 and receipt > 0):
+      return False
+    # C++ pandad stamps BOOTTIME; receipt is SubMaster MONOTONIC.
+    age = now_boot_ns - stamp
+    return 0 <= age <= ttl_ns and 0 <= now_mono_ns - receipt <= ttl_ns
+
+  def _authority(self) -> tuple[bool, bool, bool, bool, int]:
+    """Return offroad, ignition-off, voltage-safe, device-fresh, current MONO ns."""
+    clocks = self._clocks()
+    if clocks is None:
+      return False, False, False, False, self.mono_clock()
+    now_ns, boot_ns = clocks
+    sm = self.messages
+    offroad, offroad_readable = read_saved(self.params, "IsOffroad", 8)
+    shutdown, shutdown_readable = read_saved(self.params, "DoShutdown", 8)
+    manager_offroad = offroad_readable and offroad == b"1" and shutdown_readable and shutdown in (None, b"0")
+    try:
+      pandas = sm["pandaStates"]
+      ignition = tuple(bool(p.ignitionLine or p.ignitionCan) for p in pandas)
+      known_pandas = bool(pandas) and all(str(p.pandaType) != "unknown" for p in pandas)
+      ignition_off = known_pandas and not any(ignition)
+      if sm.updated["deviceState"] and int(sm.logMonoTime["deviceState"]) > self.device_after_mono_ns:
+        self.device_offset_ns = self.last_pair_offset_ns
+      evidence = ParkedEvidence(
+        manager_offroad, bool(sm.seen["deviceState"]), bool(sm.alive["deviceState"]), bool(sm.valid["deviceState"]),
+        bool(sm["deviceState"].started), int(sm.logMonoTime["deviceState"]),
+        int(sm.recv_time["deviceState"] * 1e9), self.device_offset_ns, self.device_after_mono_ns,
+        bool(sm.seen["pandaStates"]), bool(sm.alive["pandaStates"]), bool(sm.valid["pandaStates"]),
+        int(sm.logMonoTime["pandaStates"]), int(sm.recv_time["pandaStates"] * 1e9), ignition,
+      )
+      device_fresh = known_pandas and fresh_parked(evidence, now_mono_ns=now_ns, now_boot_ns=boot_ns)
+      peripheral = sm["peripheralState"]
+      power_saved = read_power(self.params)
+      cutoff_mv = effective_power(power_saved).cutoff_tenths * 100
+      voltage_ok = (power_saved.readable and power_saved.valid and self._fresh("peripheralState", now_mono_ns=now_ns,
+                     now_boot_ns=boot_ns, ttl_ns=PERIPHERAL_TTL_NS) and
+                    str(peripheral.pandaType) != "unknown" and int(peripheral.voltage) > cutoff_mv and
+                    int(sm["deviceState"].carBatteryCapacityUwh) > 0)
+      return bool(manager_offroad and not sm["deviceState"].started), ignition_off, bool(voltage_ok), device_fresh, now_ns
+    except (AttributeError, KeyError, TypeError, ValueError, OverflowError, OSError, RuntimeError):
+      return bool(manager_offroad), False, False, False, now_ns
+
+  def _sample(self, now_ns: int) -> MotionSample | None:
+    sm = self.messages
+    try:
+      if not (sm.updated["accelerometer"] and sm.seen["accelerometer"] and sm.alive["accelerometer"] and
+              sm.valid["accelerometer"]):
+        return None
+      event = sm["accelerometer"]
+      stamp = int(event.timestamp)
+      envelope = int(sm.logMonoTime["accelerometer"])
+      receipt = int(sm.recv_time["accelerometer"] * 1e9)
+      if (stamp <= max(self.last_sample_ns, self.sensor_after_mono_ns) or not 0 < stamp <= envelope <= now_ns or
+          now_ns - stamp > SAMPLE_TTL_NS or now_ns - envelope > SAMPLE_TTL_NS or
+          not 0 <= now_ns - receipt <= SAMPLE_TTL_NS):
+        return None
+      values = tuple(float(v) for v in event.acceleration.v)
+      if len(values) != 3 or any(not math.isfinite(v) for v in values):
+        return None
+      sample = MotionSample(stamp / 1e9, values)
+      self.last_sample_ns = stamp
+      return sample
+    except (AttributeError, KeyError, TypeError, ValueError, OverflowError):
+      return None
+
+  def _permitted(self, expected_raw: bytes) -> bool:
+    self.messages.update(0)
+    offroad, ignition_off, voltage_ok, fresh, now_ns = self._authority()
+    self._sample(now_ns)
+    saved = read_preferences(self.params)
+    return (saved.readable and saved.valid and saved.preferences.enabled and saved.raw == expected_raw and
+            offroad and ignition_off and voltage_ok and fresh and
+            self.policy.arm_started is not None and self.policy.last_sample_time is not None and
+            0 <= now_ns - self.last_sample_ns <= SAMPLE_TTL_NS)
+
+  def tick(self) -> Any:
+    self.messages.update(0)
+    saved = read_preferences(self.params)
+    active = (saved.readable and saved.valid and
+              saved.preferences.enabled)
+    if saved.raw != self.saved_raw:
+      self.policy = SentryPolicy(saved.preferences.settings)
+      self.saved_raw = saved.raw
+    offroad, ignition_off, voltage_ok, fresh, now_ns = self._authority()
+    sample = self._sample(now_ns) if active else None
+    decision = self.policy.update(now_ns / 1e9, Inputs(active, offroad, voltage_ok, fresh, ignition_off, sample))
+    expected_raw = saved.raw
+    if decision.event is not None and self.record is not None and expected_raw is not None and self._permitted(expected_raw):
+      try:
+        receipt = self.record(decision.event, now_ns, lambda: self._permitted(expected_raw))
+        self._recording_state("stored" if getattr(receipt, "durable", False) else "durability_unknown")
+      except Exception:
+        # No retry: storage may have published before its durability result failed.
+        self._recording_state("unavailable")
+    if self.status is not None:
+      self.status.publish(decision, saved.raw, self.recording_state)
+    return decision
+
+  def close(self) -> None:
+    for socket in getattr(self.messages, "sock", {}).values():
+      close = getattr(socket, "close", None)
+      if close is not None:
+        close()
+
+
+def main() -> None:
+  from openpilot.cereal import messaging
+  from openpilot.starpilot.sentry_mode.storage import EventStore
+
+  stop = threading.Event()
+  previous = {number: signal.getsignal(number) for number in (signal.SIGINT, signal.SIGTERM)}
+  for number in previous:
+    signal.signal(number, lambda _signal, _frame: stop.set())
+  messages = messaging.SubMaster(["accelerometer", "deviceState", "pandaStates", "peripheralState"])
+  store = EventStore()
+  from openpilot.starpilot.galaxy.camera_snapshot import CameraSnapshot, SnapshotUnavailable
+  camera = CameraSnapshot()
+  def record(kind, stamp, permitted):
+    images = {}
+    for name in ("wide", "cabin"):
+      try:
+        images[name] = camera.capture(name, permitted=permitted)
+      except SnapshotUnavailable:
+        if not permitted():
+          raise
+    return store.record(kind, stamp, permitted=permitted, images=images)
+  runtime = SentryRuntime(Params(), messages, record=record, status=RuntimeStatus())
+  try:
+    while not stop.is_set():
+      started = time.monotonic()
+      runtime.tick()
+      stop.wait(max(0.0, TICK_SECONDS - (time.monotonic() - started)))
+  finally:
+    runtime.close()
+    for number, handler in previous.items():
+      signal.signal(number, handler)
+
+
+if __name__ == "__main__":
+  main()

openpilot/starpilot/sentry_mode/status.py

@@ -0,0 +1,113 @@
+"""Expiring, namespace-bound motion-owner status; never grants runtime authority."""
+
+import hashlib
+import json
+import os
+from pathlib import Path
+import time
+import uuid
+
+from openpilot.common.hardware.hw import Paths
+from openpilot.starpilot.parked_evidence import RESUME_SKEW_NS
+from openpilot.starpilot.sentry_mode.preferences import SavedPreferences
+from openpilot.starpilot.sentry_mode.storage import EventStore, FILE_FLAGS, StorageUnavailable, _private
+from openpilot.starpilot.storage import starpilot_storage_root
+
+
+def boot_clock():
+  return time.clock_gettime_ns(getattr(time, "CLOCK_BOOTTIME", time.CLOCK_MONOTONIC))
+
+
+TTL_NS = 1_500_000_000
+STATES = {
+  "disabled": ("Off", "Motion monitoring is disabled"),
+  "disabled_onroad": ("Not armed", "The car is on or the device is shutting down"),
+  "disabled_ignition": ("Not armed", "Waiting for confirmation that the ignition is off"),
+  "low_voltage": ("Not armed", "Battery is low or its charge cannot be checked"),
+  "unavailable": ("Waiting", "Checking that the car is parked"),
+  "sensor_unavailable": ("Waiting for motion sensor", "No recent sensor readings"),
+  "arming": ("Arming", "Monitoring begins after 90 seconds parked"),
+  "armed": ("Monitoring motion", "Watching for movement while parked"),
+}
+
+
+def boot_id() -> str:
+  try:
+    return str(uuid.UUID(Path("/proc/sys/kernel/random/boot_id").read_text().strip()))
+  except (OSError, ValueError):
+    return ""
+
+
+class RuntimeStatus(EventStore):
+  def __init__(self, root: Path | None = None, *, clock=time.monotonic_ns, boot=boot_id, boot_clock=boot_clock):
+    namespace = hashlib.sha256(str(starpilot_storage_root()).encode()).hexdigest()[:24]
+    super().__init__(root if root is not None else Path(Paths.shm_path()) / f"starpilot-sentry-{namespace}")
+    self.clock, self.boot, self.boot_clock = clock, boot, boot_clock
+    self.last_published = None
+    self.last_value = None
+
+  def publish(self, decision, source: bytes | None, recording: str) -> None:
+    identity = self.boot()
+    if not identity or decision.state not in STATES:
+      return
+    current = (decision, source, recording)
+    now = self.clock()
+    if current == self.last_value and self.last_published is not None and 0 <= now - self.last_published < 500_000_000:
+      return
+    value = {"state": decision.state, "remaining": decision.seconds_remaining,
+             "source": hashlib.sha256(source or b"").hexdigest(), "boot": identity,
+             "stamp": now, "bootStamp": self.boot_clock(), "recording": recording}
+    raw = json.dumps(value, separators=(",", ":")).encode()
+    temporary = f".{uuid.uuid4().hex}"
+    try:
+      with self._directory(create=True) as directory:
+        try:
+          fd = os.open(temporary, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600, dir_fd=directory)
+          with os.fdopen(fd, "wb") as stream:
+            stream.write(raw)
+          os.replace(temporary, "status.json", src_dir_fd=directory, dst_dir_fd=directory)
+          self.last_published, self.last_value = now, current
+        finally:
+          try:
+            os.unlink(temporary, dir_fd=directory)
+          except FileNotFoundError:
+            pass
+    except (OSError, StorageUnavailable):
+      pass
+
+  def snapshot(self, saved: SavedPreferences) -> tuple[str, str]:
+    if not saved.readable or not saved.valid:
+      return "Unavailable", "Repair the saved motion settings before monitoring"
+    if not saved.preferences.enabled:
+      return STATES["disabled"]
+    try:
+      with self._directory() as directory:
+        fd = os.open("status.json", FILE_FLAGS, dir_fd=directory)
+        with os.fdopen(fd, "rb") as stream:
+          if not _private(os.fstat(stream.fileno())):
+            raise ValueError("Unsafe status")
+          raw = stream.read(1025)
+      if len(raw) > 1024:
+        raise ValueError("Oversized status")
+      value = json.loads(raw)
+      now = self.clock()
+      boot_now = self.boot_clock()
+      if (type(value) is not dict or type(value.get("stamp")) is not int or
+          type(value.get("bootStamp")) is not int or
+          not 0 <= boot_now - value["bootStamp"] <= TTL_NS or
+          abs((boot_now - value["bootStamp"]) - (now - value["stamp"])) > RESUME_SKEW_NS or
+          not 0 <= now - value["stamp"] <= TTL_NS or not self.boot() or value.get("boot") != self.boot() or
+          value.get("source") != hashlib.sha256(saved.raw or b"").hexdigest() or
+          value.get("state") not in STATES):
+        raise ValueError("Stale status")
+      label, reason = STATES[value["state"]]
+      if value["state"] == "arming":
+        remaining = value.get("remaining")
+        if type(remaining) is not int or not 0 <= remaining <= 90:
+          raise ValueError("Invalid countdown")
+        label = f"Arming · {remaining}s"
+      if value.get("recording") in ("unavailable", "durability_unknown"):
+        reason += "; motion events could not be saved reliably"
+      return label, reason
+    except (OSError, StorageUnavailable, ValueError, TypeError, KeyError, RecursionError):
+      return "Waiting for monitor", "The monitor has not responded recently"

openpilot/starpilot/sentry_mode/storage.py

@@ -0,0 +1,285 @@
+"""Bounded durable Sentry events and captured camera evidence."""
+
+from collections.abc import Callable
+from contextlib import contextmanager
+from dataclasses import dataclass
+import fcntl
+import json
+import os
+from pathlib import Path
+import re
+import stat
+import time
+import uuid
+
+from openpilot.starpilot.storage import starpilot_storage_root
+
+
+MAX_EVENTS = 512
+MAX_EVENT_BYTES = 512
+ID = re.compile(r"[0-9a-f]{32}\Z")
+FIELDS = {"version", "eventId", "sessionId", "kind", "monoTimeNs", "wallTimeNs"}
+DIR_FLAGS = os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW
+FILE_FLAGS = os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK
+
+
+class StorageUnavailable(Exception):
+  """No event was published by this call; recording cannot continue."""
+
+
+@dataclass(frozen=True)
+class RecordReceipt:
+  event_id: str
+  # False means published but persistence across power loss was not confirmed.
+  durable: bool
+
+
+def _private(info: os.stat_result, *, directory: bool = False) -> bool:
+  return (stat.S_ISDIR(info.st_mode) if directory else stat.S_ISREG(info.st_mode) and info.st_nlink == 1) and \
+    info.st_uid == os.geteuid() and info.st_mode & 0o077 == 0
+
+
+def _valid(value: object, event_id: str) -> bool:
+  if type(value) is not dict:
+    return False
+  if any(type(key) is not str for key in value):
+    return False
+  fields = {key: item for key, item in value.items() if type(key) is str}
+  if set(fields) != FIELDS or ID.fullmatch(event_id) is None:
+    return False
+  version = fields["version"]
+  session_id = fields["sessionId"]
+  mono_time = fields["monoTimeNs"]
+  wall_time = fields["wallTimeNs"]
+  return (type(version) is int and version == 1 and
+          fields["eventId"] == event_id and
+          type(session_id) is str and ID.fullmatch(session_id) is not None and
+          fields["kind"] in ("warning", "alarm") and
+          type(mono_time) is int and 0 <= mono_time < 2**63 and
+          type(wall_time) is int and 0 <= wall_time < 2**63)
+
+
+def _unique(pairs: list[tuple[str, object]]) -> dict:
+  result: dict = {}
+  for key, value in pairs:
+    if key in result:
+      raise ValueError("duplicate event field")
+    result[key] = value
+  return result
+
+
+class EventStore:
+  def __init__(self, root: Path | None = None):
+    self.root = root if root is not None else starpilot_storage_root() / "sentry/events"
+    self.session_id = uuid.uuid4().hex
+
+  def _open_directory(self, *, create: bool = False) -> int:
+    # Traverse from an already-open filesystem root. mkdir(parents=True) would
+    # follow a replaced ancestor before the final O_NOFOLLOW check.
+    if not self.root.is_absolute() or ".." in self.root.parts:
+      raise StorageUnavailable("Event directory path is unavailable")
+    fd = os.open("/", DIR_FLAGS)
+    try:
+      for component in self.root.parts[1:]:
+        if create:
+          try:
+            os.mkdir(component, mode=0o700, dir_fd=fd)
+          except FileExistsError:
+            pass
+        child = os.open(component, DIR_FLAGS, dir_fd=fd)
+        os.close(fd)
+        fd = child
+      return fd
+    except BaseException:
+      os.close(fd)
+      raise
+
+  def _same_directory(self, fd: int) -> bool:
+    try:
+      current = self._open_directory()
+      try:
+        opened_info, current_info = os.fstat(fd), os.fstat(current)
+        return (opened_info.st_dev, opened_info.st_ino) == (current_info.st_dev, current_info.st_ino)
+      finally:
+        os.close(current)
+    except OSError:
+      return False
+
+  @contextmanager
+  def _directory(self, *, create: bool = False):
+    fd = self._open_directory(create=create)
+    try:
+      info = os.fstat(fd)
+      if not _private(info, directory=True) or not self._same_directory(fd):
+        raise StorageUnavailable("Event directory is unavailable")
+      yield fd
+    finally:
+      os.close(fd)
+
+  def record(self, kind: str, mono_time_ns: int, *, permitted: Callable[[], bool], images: dict | None = None) -> RecordReceipt:
+    event_id = uuid.uuid4().hex
+    value = {"version": 1, "eventId": event_id, "sessionId": self.session_id, "kind": kind,
+             "monoTimeNs": mono_time_ns, "wallTimeNs": time.time_ns()}
+    if not _valid(value, event_id):
+      raise ValueError("Invalid Sentry event")
+    raw = json.dumps(value, sort_keys=True, separators=(",", ":"), allow_nan=False).encode()
+    if len(raw) > MAX_EVENT_BYTES:
+      raise ValueError("Sentry event exceeds its storage limit")
+    if not permitted():
+      raise StorageUnavailable("Sentry is no longer armed")
+    published = False
+    try:
+      with self._directory(create=True) as directory:
+        lock = os.open(".lock", os.O_CREAT | os.O_RDWR | os.O_NOFOLLOW | os.O_NONBLOCK, 0o600, dir_fd=directory)
+        try:
+          if not _private(os.fstat(lock)):
+            raise StorageUnavailable("Event lock is unavailable")
+          fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
+          # Unknown entries and interrupted writes also consume capacity. Never
+          # delete evidence, or scan an unbounded directory to admit a new event.
+          count = 0
+          with os.scandir(directory) as entries:
+            for entry in entries:
+              if ID.fullmatch(entry.name) and entry.is_dir(follow_symlinks=False):
+                try:
+                  metadata = os.stat(entry.name + ".json", dir_fd=directory, follow_symlinks=False)
+                  if _private(metadata):
+                    continue  # Evidence directory and published metadata are one event.
+                except OSError:
+                  pass  # Interrupted evidence still consumes capacity.
+              if entry.name != ".lock":
+                count += 1
+                if count >= MAX_EVENTS:
+                  raise StorageUnavailable("Event storage is full")
+          for camera, body in (images or {}).items():
+            if camera not in ("wide", "cabin") or not isinstance(body, bytes) or not 4 <= len(body) <= 1_000_000 or not body.startswith(b"\xff\xd8") or not body.endswith(b"\xff\xd9"):
+              raise ValueError("Invalid event image")
+          if images:
+            os.mkdir(event_id, mode=0o700, dir_fd=directory)
+            image_directory = os.open(event_id, DIR_FLAGS, dir_fd=directory)
+            try:
+              for camera, body in images.items():
+                fd = os.open(camera + ".jpg", os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600, dir_fd=image_directory)
+                with os.fdopen(fd, "wb") as output:
+                  output.write(body)
+                  output.flush()
+                  os.fsync(output.fileno())
+              os.fsync(image_directory)
+            finally:
+              os.close(image_directory)
+          temporary = f".{event_id}.tmp"
+          fd = os.open(temporary, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600, dir_fd=directory)
+          try:
+            with os.fdopen(fd, "wb") as output:
+              output.write(raw)
+              output.flush()
+              os.fsync(output.fileno())
+            if not self._same_directory(directory) or not permitted():
+              raise StorageUnavailable("Sentry authority or event directory changed")
+            # An atomic no-replace publication: a coincident ID cannot overwrite
+            # an existing event, including an unexpected symlink.
+            os.link(temporary, event_id + ".json", src_dir_fd=directory, dst_dir_fd=directory, follow_symlinks=False)
+            published = True
+          finally:
+            os.unlink(temporary, dir_fd=directory)
+          os.fsync(directory)
+        finally:
+          os.close(lock)
+    except (OSError, StorageUnavailable) as error:
+      if published:
+        return RecordReceipt(event_id, durable=False)
+      raise StorageUnavailable(str(error)) from error
+    return RecordReceipt(event_id, durable=True)
+
+  def snapshot(self) -> dict:
+    result: dict = {"version": 1, "events": [], "incomplete": False, "capacity": MAX_EVENTS}
+    try:
+      with self._directory() as directory:
+        with os.scandir(directory) as entries:
+          count = 0
+          for entry in entries:
+            if entry.name == ".lock" or (ID.fullmatch(entry.name) and entry.is_dir(follow_symlinks=False)):
+              continue
+            count += 1
+            if count > MAX_EVENTS:
+              result["incomplete"] = True
+              break
+            event_id = entry.name.removesuffix(".json")
+            if entry.name != event_id + ".json" or ID.fullmatch(event_id) is None:
+              result["incomplete"] = True
+              continue
+            try:
+              fd = os.open(entry.name, FILE_FLAGS, dir_fd=directory)
+              try:
+                if not _private(os.fstat(fd)):
+                  raise ValueError("Invalid event file")
+                raw = os.read(fd, MAX_EVENT_BYTES + 1)
+              finally:
+                os.close(fd)
+              if len(raw) > MAX_EVENT_BYTES:
+                raise ValueError("Event too large")
+              value = json.loads(raw, object_pairs_hook=_unique)
+              if not _valid(value, event_id):
+                raise ValueError("Invalid event metadata")
+              result["events"].append(value)
+            except (OSError, ValueError, TypeError, RecursionError):
+              result["incomplete"] = True
+        if not self._same_directory(directory):
+          raise StorageUnavailable("Event directory changed")
+    except FileNotFoundError:
+      return result
+    except OSError as error:
+      raise StorageUnavailable("Event storage unavailable") from error
+    result["events"].sort(key=lambda event: (event["wallTimeNs"], event["eventId"]), reverse=True)
+    return result
+
+  def image(self, event_id: str, camera: str) -> bytes:
+    if ID.fullmatch(event_id) is None or camera not in ("wide", "cabin"):
+      raise StorageUnavailable("Unknown event image")
+    with self._directory() as directory:
+      # Only published events can expose images.
+      event_fd = os.open(event_id + ".json", FILE_FLAGS, dir_fd=directory)
+      try:
+        if not _private(os.fstat(event_fd)):
+          raise StorageUnavailable("Invalid event")
+      finally:
+        os.close(event_fd)
+      images = os.open(event_id, DIR_FLAGS, dir_fd=directory)
+      try:
+        if not _private(os.fstat(images), directory=True):
+          raise StorageUnavailable("Invalid image directory")
+        fd = os.open(camera + ".jpg", FILE_FLAGS, dir_fd=images)
+        try:
+          if not _private(os.fstat(fd)):
+            raise StorageUnavailable("Invalid image")
+          body = os.read(fd, 1_000_001)
+        finally:
+          os.close(fd)
+      finally:
+        os.close(images)
+    if not 4 <= len(body) <= 1_000_000 or not body.startswith(b"\xff\xd8") or not body.endswith(b"\xff\xd9"):
+      raise StorageUnavailable("Invalid image")
+    return body
+
+  def images(self, event_id: str) -> list[str]:
+    if ID.fullmatch(event_id) is None:
+      return []
+    try:
+      with self._directory() as directory:
+        images = os.open(event_id, DIR_FLAGS, dir_fd=directory)
+        try:
+          if not _private(os.fstat(images), directory=True):
+            return []
+          result = []
+          for camera in ("wide", "cabin"):
+            try:
+              info = os.stat(camera + ".jpg", dir_fd=images, follow_symlinks=False)
+              if _private(info) and 4 <= info.st_size <= 1_000_000:
+                result.append(camera)
+            except OSError:
+              pass
+          return result
+        finally:
+          os.close(images)
+    except (OSError, StorageUnavailable):
+      return []

openpilot/starpilot/sentry_mode/tests/test_manager.py

@@ -0,0 +1,37 @@
+import os
+from pathlib import Path
+import tempfile
+import unittest
+from unittest.mock import patch
+
+from openpilot.common.params import Params
+from openpilot.starpilot.sentry_mode.preferences import KEY, Preferences, encode
+from openpilot.system.manager import process_config
+
+
+class TestSentryManagerPredicate(unittest.TestCase):
+  def test_device_and_pc_matrix_requires_enabled_and_valid_saved_document(self):
+    with tempfile.TemporaryDirectory() as directory, patch.dict(os.environ, PARAMS_ROOT=directory, OPENPILOT_PREFIX="sentrymanager",
+                                                                  STARPILOT_SENTRY_DEVELOPMENT="0"):
+      params = Params()
+      process = process_config.managed_processes["sentry_motion"]
+      sensor = process_config.managed_processes["sensord"]
+      self.assertEqual(process.enabled, not process_config.PC)
+      self.assertEqual(sensor.enabled, not process_config.PC)
+      with patch.dict(os.environ, STARPILOT_SENTRY_DEVELOPMENT="1"):
+        self.assertFalse(process_config.sentry_motion(False, params, object()))
+        self.assertFalse(process_config.sensord_run(False, params, object()))
+        self.assertTrue(process_config.sensord_run(True, params, object()))
+        source = Path(params.get_param_path(KEY))
+        source.write_bytes(b"invalid")
+        self.assertFalse(process_config.sentry_motion(False, params, object()))
+        source.write_bytes(encode(Preferences(True)))
+        self.assertTrue(process_config.sentry_motion(False, params, object()))
+        self.assertTrue(process_config.sensord_run(False, params, object()))
+        self.assertFalse(process_config.sentry_motion(True, params, object()))
+      self.assertTrue(process_config.sentry_motion(False, params, object()))
+      self.assertTrue(process_config.sensord_run(False, params, object()))
+
+
+if __name__ == "__main__":
+  unittest.main()

openpilot/starpilot/sentry_mode/tests/test_notifications.py

@@ -0,0 +1,214 @@
+"""Private credentials, actual HTTP contracts, durable retries and RFC push crypto."""
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+import json
+import struct
+import threading
+from types import SimpleNamespace
+from urllib.parse import parse_qs
+
+from cryptography.hazmat.primitives import hashes
+from cryptography.hazmat.primitives.asymmetric import ec, utils
+from cryptography.hazmat.primitives.ciphers.aead import AESGCM
+from cryptography.hazmat.primitives.kdf.hkdf import HKDF
+import pytest
+
+from openpilot.starpilot.sentry_mode.notifications import NotificationOwner, NotificationUnavailable, CHANNELS
+from openpilot.starpilot.sentry_mode import web_push
+
+
+def store(events=()):
+  return SimpleNamespace(snapshot=lambda: {'events': list(events)})
+
+
+def configure(owner, channel, url='https://example.test/private-secret', enabled=True):
+  return owner.action({'action': 'configure', 'channel': channel, 'enabled': enabled, 'url': url, 'token': 'private-token'})
+
+
+def test_private_configuration_redaction_validation_and_exclusive_owner(tmp_path):
+  owner = NotificationOwner(tmp_path / 'notifications', store())
+  try:
+    snapshot = configure(owner, 'ntfy')
+    assert snapshot['channels']['ntfy']['enabled'] and snapshot['channels']['ntfy']['configured']
+    assert 'private-secret' not in json.dumps(snapshot) and 'private-token' not in json.dumps(snapshot)
+    assert (tmp_path / 'notifications/state.json').stat().st_mode & 0o777 == 0o600
+    duplicate = NotificationOwner(tmp_path / 'notifications', store())
+    with pytest.raises(NotificationUnavailable):
+      duplicate.snapshot()
+    duplicate.close()
+    for url in ('http://example.test/a', 'https://u:p@example.test', 'https://example.test/#x', 'https://example.test/\n'):
+      with pytest.raises(ValueError):
+        configure(owner, 'webhook', url)
+    before = owner.snapshot()
+    with pytest.raises(PermissionError):
+      configure_kwargs = {'action': 'forget', 'channel': 'ntfy'}
+      owner.action(configure_kwargs, permitted=lambda: False)
+    assert owner.snapshot() == before
+  finally:
+    owner.close()
+
+
+def test_event_scan_durable_dedupe_retry_and_disable(tmp_path):
+  clock = [100.]
+  calls = []
+  outcomes = [503, 202]
+  events = []
+  def transport(*args):
+    calls.append(args)
+    return outcomes.pop(0)
+  root = tmp_path / 'notifications'
+  owner = NotificationOwner(root, store(events), clock=lambda: clock[0], transport=transport)
+  configure(owner, 'webhook')
+  events.append({'eventId': 'a'*32, 'kind': 'alarm', 'wallTimeNs': int(101e9)})
+  clock[0] = 101.
+  owner.tick()
+  assert len(calls) == 1 and owner.snapshot()['channels']['webhook']['lastError'] == 'http'
+  owner.close()
+  owner = NotificationOwner(root, store(events), clock=lambda: clock[0], transport=transport)
+  owner.tick()
+  assert len(calls) == 1
+  clock[0] = 132.
+  owner.tick()
+  assert len(calls) == 2 and calls[0][4] == calls[1][4]
+  assert owner.snapshot()['channels']['webhook']['lastState'] == 'sent'
+  owner.tick()
+  assert len(calls) == 2
+  configure(owner, 'webhook', enabled=False)
+  events.append({'eventId': 'b'*32, 'kind': 'warning', 'wallTimeNs': int(133e9)})
+  clock[0] = 133.
+  owner.tick()
+  assert len(calls) == 2
+  owner.close()
+
+
+def test_real_http_contracts_and_independent_channels(tmp_path):
+  received = []
+  class Handler(BaseHTTPRequestHandler):
+    def do_POST(self):
+      received.append((self.path, self.headers, self.rfile.read(int(self.headers['Content-Length']))))
+      self.send_response(200)
+      self.send_header('Content-Length', '100000000')
+      self.end_headers()
+      self.wfile.write(b'x')  # Sender must read status only, not this unbounded/incomplete body.
+    def log_message(self, *args):
+      pass
+  server = ThreadingHTTPServer(('127.0.0.1', 0), Handler)
+  thread = threading.Thread(target=server.serve_forever, daemon=True)
+  thread.start()
+  owner = NotificationOwner(tmp_path / 'notifications', store(), allow_http=True)
+  try:
+    for name in ('discord', 'webhook', 'ntfy'):
+      configure(owner, name, f'http://127.0.0.1:{server.server_port}/{name}')
+      owner.action({'action': 'test', 'channel': name})
+    owner.action({'action': 'pushKey'})
+    receiver = ec.generate_private_key(ec.SECP256R1())
+    owner.action({'action': 'subscribe', 'subscription': {'endpoint': f'http://127.0.0.1:{server.server_port}/push',
+                  'keys': {'p256dh': web_push.b64(web_push.public(receiver)), 'auth': web_push.b64(b'0123456789abcdef')}}})
+    owner.action({'action': 'configure', 'channel': 'webPush', 'enabled': True, 'url': '', 'token': ''})
+    owner.action({'action': 'test', 'channel': 'webPush'})
+    for _ in range(4):
+      owner.tick()
+    assert len(received) == 4
+    paths = {path: (headers, body) for path, headers, body in received}
+    assert json.loads(paths['/discord'][1])['content'].startswith('StarPilot Sentry Mode:')
+    assert json.loads(parse_qs(paths['/webhook'][1].decode())['event'][0])['kind'] == 'test'
+    assert paths['/ntfy'][0]['Priority'] == 'urgent'
+    assert paths['/ntfy'][1].startswith(b'This is a test')
+    for path, (headers, _) in paths.items():
+      assert len(headers['Idempotency-Key']) == 64
+      if path != '/push':
+        assert headers['Authorization'] == 'Bearer private-token'
+    assert paths['/push'][0]['Authorization'].startswith('vapid t=')
+    assert paths['/push'][0]['Content-Encoding'] == 'aes128gcm'
+    assert len(paths['/push'][1]) > 100 and b'This is a test' not in paths['/push'][1]
+    assert all(owner.snapshot()['channels'][name]['lastState'] == 'sent' for name in CHANNELS)
+  finally:
+    owner.close()
+    server.shutdown()
+    server.server_close()
+    thread.join(timeout=5)
+
+
+def test_push_encrypts_decryptable_payload_and_valid_signed_vapid():
+  receiver = ec.generate_private_key(ec.SECP256R1())
+  auth = b'0123456789abcdef'
+  subscription = {'endpoint': 'https://push.example.test/private', 'keys': {'p256dh': web_push.b64(web_push.public(receiver)), 'auth': web_push.b64(auth)}}
+  pem = web_push.new_key()
+  message = {'eventId': 'a'*32, 'body': 'Parked motion detected'}
+  body, headers = web_push.request(subscription, message, pem, now=100)
+  salt, rs, key_length = body[:16], struct.unpack('!I', body[16:20])[0], body[20]
+  assert rs == 4096 and key_length == 65
+  sender_public = body[21:86]
+  shared = receiver.exchange(ec.ECDH(), ec.EllipticCurvePublicKey.from_encoded_point(ec.SECP256R1(), sender_public))
+  ikm = HKDF(algorithm=hashes.SHA256(), length=32, salt=auth, info=b'WebPush: info\x00' + web_push.public(receiver) + sender_public).derive(shared)
+  cek = HKDF(algorithm=hashes.SHA256(), length=16, salt=salt, info=b'Content-Encoding: aes128gcm\x00').derive(ikm)
+  nonce = HKDF(algorithm=hashes.SHA256(), length=12, salt=salt, info=b'Content-Encoding: nonce\x00').derive(ikm)
+  plaintext = AESGCM(cek).decrypt(nonce, body[86:], None)
+  assert plaintext[-1] == 2 and json.loads(plaintext[:-1]) == message
+  assert b'Parked motion' not in body
+  token = headers['Authorization'].split('t=', 1)[1].split(',', 1)[0]
+  head, claims, sig = token.split('.')
+  claim_values = json.loads(web_push.unb64(claims))
+  assert claim_values['aud'] == 'https://push.example.test' and claim_values['exp'] == 43300
+  raw_signature = web_push.unb64(sig)
+  der = utils.encode_dss_signature(int.from_bytes(raw_signature[:32], 'big'), int.from_bytes(raw_signature[32:], 'big'))
+  web_push.load_key(pem).public_key().verify(der, (head + '.' + claims).encode(), ec.ECDSA(hashes.SHA256()))
+
+
+def test_bounded_retry_and_expired_push_subscription(tmp_path):
+  clock = [100.]
+  calls = []
+  owner = NotificationOwner(tmp_path / 'notifications', store(), clock=lambda: clock[0], transport=lambda *args: calls.append(args) or 503)
+  try:
+    configure(owner, 'ntfy')
+    owner.action({'action': 'test', 'channel': 'ntfy'})
+    for stamp in (100., 130., 250., 850., 2000.):
+      clock[0] = stamp
+      owner.tick()
+    assert len(calls) == 4 and owner.snapshot()['channels']['ntfy']['lastState'] == 'failed'
+    owner.action({'action': 'pushKey'})
+    receiver = ec.generate_private_key(ec.SECP256R1())
+    owner.action({'action': 'subscribe', 'subscription': {'endpoint': 'https://push.example.test/private',
+                  'keys': {'p256dh': web_push.b64(web_push.public(receiver)), 'auth': web_push.b64(b'0123456789abcdef')}}})
+    owner.action({'action': 'configure', 'channel': 'webPush', 'enabled': True, 'url': '', 'token': ''})
+    owner.action({'action': 'test', 'channel': 'webPush'})
+    owner.transport = lambda *args: 410
+    owner.tick()
+    assert owner.snapshot()['subscriptionCount'] == 0
+    assert owner.snapshot()['channels']['webPush']['lastError'] == 'expired'
+  finally:
+    owner.close()
+
+
+def test_published_rfc8291_section5_exact_ciphertext():
+  # Independent published wire vector, https://www.rfc-editor.org/rfc/rfc8291#section-5
+  subscription = {'endpoint': 'https://push.example.net/push/test', 'keys': {
+    'auth': 'BTBZMqHH6r4Tts7J_aSIgg',
+    'p256dh': 'BCVxsr7N_eNgVRqvHtD0zTZsEc6-VV-JvLexhqUzORcxaOzi6-AYWXvTBHm4bjyPjs7Vd8pZGH6SRpkNtoIAiw4'}}
+  sender = ec.derive_private_key(int.from_bytes(web_push.unb64('yfWPiYE-n46HLnH0KqZOF1fJJU3MYrct3AELtAQ-oRw'), 'big'), ec.SECP256R1())
+  expected = web_push.unb64('DGv6ra1nlYgDCS1FRnbzlwAAEABBBP4z9KsN6nGRTbVYI_c7VJSPQTBtkgcy27ml' +
+                           'mlMoZIIgDll6e3vCYLocInmYWAmS6TlzAC8wEqKK6PBru3jl7A_yl95bQpu6cVPT' +
+                           'pK4Mqgkf1CXztLVBSt2Ks3oZwbuwXPXLWyouBWLVWGNWQexSgSxsj_Qulcy4a-fN')
+  actual = web_push.encrypt(subscription, b'When I grow up, I want to be a watermelon', sender=sender, salt=expected[:16])
+  assert actual == expected
+
+
+@pytest.mark.parametrize('broken', [None, [], {}, 'bad', float('nan')])
+def test_corrupt_saved_generation_and_timestamp_never_repaired_or_sent(tmp_path, broken):
+  root = tmp_path / 'notifications'
+  owner = NotificationOwner(root, store())
+  configure(owner, 'webhook')
+  owner.close()
+  path = root / 'state.json'
+  value = json.loads(path.read_text())
+  value['channels']['webhook']['generation'] = broken
+  original = json.dumps(value).encode()
+  path.write_bytes(original)
+  calls = []
+  owner = NotificationOwner(root, store(), transport=lambda *args: calls.append(args) or 202)
+  try:
+    with pytest.raises(NotificationUnavailable):
+      owner.snapshot()
+    owner.tick()
+    assert calls == [] and path.read_bytes() == original
+  finally:
+    owner.close()

openpilot/starpilot/sentry_mode/tests/test_policy.py

@@ -0,0 +1,140 @@
+"""Frozen numeric motion behavior and fail-closed parked policy boundaries."""
+
+from unittest.mock import Mock
+
+import pytest
+
+from openpilot.starpilot.sentry_mode.policy import (
+  Inputs, MotionSample, SentryPolicy, Settings,
+)
+
+
+def active(sample: MotionSample | None = None) -> Inputs:
+  return Inputs(enabled=True, offroad=True, voltage_ok=True, device_fresh=True, ignition_off=True, sample=sample)
+
+
+def motion(now: float, z: float) -> MotionSample:
+  return MotionSample(now, (0.0, 0.0, z))
+
+
+def fresh_through(policy: SentryPolicy, start: float, end: float, z: float = 9.8):
+  result = None
+  for tenth in range(round(start * 10) + 1, round(end * 10) + 1):
+    now = tenth / 10
+    result = policy.update(now, active(motion(now, z)))
+  return result
+
+
+def armed(policy: SentryPolicy) -> None:
+  assert policy.update(0.0, active(motion(0.0, 9.8))).state == "arming"
+  assert fresh_through(policy, 0.0, 89.9).state == "arming"
+  assert policy.update(90.0, active(motion(90.0, 9.8))).state == "armed"
+
+
+def test_frozen_arming_warning_and_alarm_numeric_sequence():
+  policy = SentryPolicy(Settings(sensitivity=0.1, warning_time_seconds=0.3))
+  armed(policy)
+  assert policy.settings.warning_trigger_count == 3
+  assert policy.update(90.1, active(motion(90.1, 10.0))).event is None
+  assert policy.update(90.2, active(motion(90.2, 9.8))).event is None
+  assert policy.update(90.3, active(motion(90.3, 10.0))).event == "warning"
+  assert policy.update(90.4, active(motion(90.4, 9.8))).event is None
+  # The frozen detector requires strictly more than 25 changes and 30 s from
+  # the first change. Feed 10 Hz fresh stable samples while that time elapses.
+  for index in range(5, 26):
+    now = 90.0 + index * 0.1
+    assert policy.update(now, active(motion(now, 10.0 if index % 2 else 9.8))).event is None
+  assert policy.trigger_count == 25
+  assert fresh_through(policy, 92.5, 120.0, 10.0).event is None
+  assert policy.update(120.1, active(motion(120.1, 9.8))).event == "alarm"
+  assert policy.update(120.2, active(motion(120.2, 10.0))).event is None
+
+
+def test_duplicate_stale_invalid_and_out_of_order_samples_do_not_accumulate():
+  policy = SentryPolicy(Settings(sensitivity=0.1, warning_time_seconds=0.2))
+  armed(policy)
+  assert policy.update(90.1, active(motion(90.1, 10.0))).event is None
+  assert policy.update(90.2, active(motion(90.1, 9.8))).state == "sensor_unavailable"
+  assert policy.trigger_count == 0
+  assert policy.update(90.3, active(motion(90.1, 9.8))).state == "sensor_unavailable"
+  assert policy.update(90.4, active(motion(90.4, 9.8))).state == "arming"
+  assert policy.update(90.5, active(motion(90.3, 10.0))).state == "sensor_unavailable"
+  assert policy.update(90.6, active(motion(90.6, 9.8))).state == "arming"
+
+
+def test_missing_sensor_during_arming_and_after_armed_requires_full_rearm():
+  policy = SentryPolicy()
+  assert policy.update(0.0, active()).state == "sensor_unavailable"
+  assert policy.update(0.1, active(motion(0.1, 9.8))).state == "arming"
+  assert fresh_through(policy, 0.1, 90.0).state == "arming"
+  assert policy.update(90.1, active()).state == "sensor_unavailable"
+  assert policy.update(90.2, active(motion(90.2, 9.8))).state == "arming"
+  assert fresh_through(policy, 90.2, 180.3).state == "armed"
+  assert policy.update(180.4, active()).state == "sensor_unavailable"
+  assert policy.update(180.5, active(motion(180.5, 9.8))).state == "arming"
+
+
+def test_disabled_onroad_low_voltage_and_lost_device_rearm_for_full_90_seconds():
+  for stopped, expected in ((Inputs(False, True, True, True, True), "disabled"),
+                            (Inputs(True, False, True, True, True), "disabled_onroad"),
+                            (Inputs(True, True, False, True, True), "low_voltage"),
+                            (Inputs(True, True, True, False, True), "unavailable"),
+                            (Inputs(True, True, True, True, False), "disabled_ignition")):
+    policy = SentryPolicy()
+    armed(policy)
+    assert policy.update(90.1, active(motion(90.1, 10.0))).state == "armed"
+    assert policy.update(90.2, stopped).state == expected
+    assert policy.update(90.3, active(motion(90.3, 9.8))).state == "arming"
+    assert fresh_through(policy, 90.3, 180.2).state == "arming"
+    assert fresh_through(policy, 180.2, 180.4).state == "armed"
+    assert policy.trigger_count == 0
+
+
+def test_monotonic_reversal_and_motion_window_expiry_fail_closed():
+  policy = SentryPolicy(Settings(sensitivity=0.1, warning_time_seconds=0.2))
+  armed(policy)
+  policy.update(90.1, active(motion(90.1, 10.0)))
+  assert policy.update(60.0, active(motion(60.0, 9.8))).state == "unavailable"
+  assert policy.update(60.1, active(motion(60.1, 9.8))).state == "arming"
+  assert fresh_through(policy, 60.1, 150.0).state == "arming"
+  assert fresh_through(policy, 150.0, 150.2).state == "armed"
+  assert policy.update(150.3, active(motion(150.3, 10.0))).event is None
+  assert fresh_through(policy, 150.3, 210.2, 10.0).event is None
+  assert policy.update(210.3, active(motion(210.3, 9.8))).event is None
+  assert policy.trigger_count == 1
+
+
+def test_two_samples_90_seconds_apart_never_arm():
+  policy = SentryPolicy()
+  assert policy.update(0.0, active(motion(0.0, 9.8))).state == "arming"
+  assert policy.update(90.0, active(motion(90.0, 9.8))).state == "unavailable"
+  assert policy.update(90.1, active(motion(90.1, 9.8))).state == "arming"
+  assert fresh_through(policy, 90.1, 180.2).state == "armed"
+
+
+@pytest.mark.parametrize("settings", [
+  {"sensitivity": 0.004}, {"sensitivity": float("nan")},
+  {"sensitivity": True}, {"sensitivity": "0.04"},
+  {"warning_time_seconds": 10.1}, {"warning_time_seconds": float("inf")},
+  {"warning_time_seconds": False},
+])
+def test_settings_reject_invalid_values(settings):
+  with pytest.raises(ValueError):
+    Settings(**settings)
+
+
+@pytest.mark.parametrize("sample", [
+  (float("nan"), (0.0, 0.0, 9.8)),
+  (1.0, (0.0, float("inf"), 9.8)),
+  (1.0, (1e308, 0.0, 9.8)),
+  (1.0, (True, 0.0, 9.8)),
+  (1.0, (0.0, 9.8)),
+])
+def test_sample_requires_finite_time_and_three_axes(sample):
+  with pytest.raises(ValueError):
+    MotionSample(*sample)
+
+
+def test_authority_inputs_reject_truthy_strings():
+  with pytest.raises(ValueError):
+    Mock(wraps=Inputs)("off", True, True, True, True)

openpilot/starpilot/sentry_mode/tests/test_preferences.py

@@ -0,0 +1,52 @@
+import os
+from pathlib import Path
+import tempfile
+import unittest
+from unittest.mock import patch
+
+from openpilot.common.params import Params
+from openpilot.starpilot.sentry_mode.policy import Settings
+from openpilot.starpilot.sentry_mode.preferences import KEY, Preferences, decode, encode, enabled, read_preferences
+
+
+class TestPreferences(unittest.TestCase):
+  def test_codec_requires_exact_finite_document(self):
+    value = Preferences(True, Settings(0.04, 1.0))
+    self.assertEqual(decode(encode(value)), value)
+    corrupt = (
+      b'{}', b'{"version":1,"enabled":true,"sensitivity":0.04,"warningTimeSeconds":1,"extra":0}',
+      b'{"version":1,"enabled":true,"enabled":false,"sensitivity":0.04,"warningTimeSeconds":1}',
+      b'{"version":1,"enabled":true,"sensitivity":NaN,"warningTimeSeconds":1}',
+      b'{"version":1,"enabled":true,"sensitivity":0.04,"warningTimeSeconds":1e400}',
+      b'{"version":1,"enabled":1,"sensitivity":0.04,"warningTimeSeconds":1}',
+      b'[' * 250 + b'0' + b']' * 250,
+    )
+    for raw in corrupt:
+      with self.subTest(raw=raw[:30]):
+        self.assertIsNone(decode(raw))
+
+  def test_actual_params_absent_corrupt_and_oversized_are_off_without_rewrite(self):
+    with tempfile.TemporaryDirectory() as directory, patch.dict(os.environ, PARAMS_ROOT=directory, OPENPILOT_PREFIX="sentrytest"):
+      params = Params()
+      self.assertFalse(enabled(params))
+      raw = encode(Preferences(True))
+      Path(params.get_param_path(KEY)).write_bytes(raw)
+      self.assertTrue(enabled(params))
+      for invalid in (b'{"version":2}', b'x' * 513):
+        Path(params.get_param_path(KEY)).write_bytes(invalid)
+        saved = read_preferences(params)
+        self.assertFalse(saved.valid)
+        self.assertFalse(enabled(params))
+        self.assertEqual(Path(params.get_param_path(KEY)).read_bytes(), invalid)
+
+  def test_unreadable_sources_do_not_arm(self):
+    with tempfile.TemporaryDirectory() as directory, patch.dict(os.environ, PARAMS_ROOT=directory, OPENPILOT_PREFIX="sentrytest"):
+      params = Params()
+      source = Path(params.get_param_path(KEY))
+      source.symlink_to(Path(directory) / "elsewhere")
+      self.assertFalse(read_preferences(params).readable)
+      self.assertFalse(enabled(params))
+
+
+if __name__ == "__main__":
+  unittest.main()

openpilot/starpilot/sentry_mode/tests/test_runtime.py

@@ -0,0 +1,215 @@
+import os
+import signal
+from contextlib import ExitStack
+from pathlib import Path
+from types import SimpleNamespace as NS
+import tempfile
+import unittest
+from unittest.mock import Mock, patch
+
+from openpilot.cereal import log, messaging
+from openpilot.common.params import Params
+from openpilot.starpilot.sentry_mode.preferences import KEY, Preferences, encode
+from openpilot.starpilot.sentry_mode.runtime import SentryRuntime
+from openpilot.starpilot.sentry_mode import runtime as runtime_module
+
+
+class Clock:
+  def __init__(self):
+    self.mono = 1_000_000_000_000
+    self.offset = 50_000_000_000
+
+  def boot(self):
+    return self.mono + self.offset
+
+
+class Messages:
+  def __init__(self, clock):
+    self.clock = clock
+    self.messages = {}
+    names = ("deviceState", "pandaStates", "peripheralState", "accelerometer")
+    self.seen = dict.fromkeys(names, True)
+    self.alive = dict.fromkeys(names, True)
+    self.valid = dict.fromkeys(names, True)
+    self.updated = dict.fromkeys(names, True)
+    self.logMonoTime = dict.fromkeys(names, 0)
+    self.recv_time = dict.fromkeys(names, 0.0)
+    self.sock = {}
+
+  def __getitem__(self, name):
+    return self.messages[name]
+
+  def update(self, _timeout):
+    pass
+
+  def set_sources(self, *, sample_stamp=None, acceleration=(0.0, 0.0, 9.81)):
+    now = self.clock.mono
+    for name in self.logMonoTime:
+      self.recv_time[name] = (now - 10_000_000) / 1e9
+      self.logMonoTime[name] = now - 10_000_000 if name in ("deviceState", "accelerometer") else self.clock.boot() - 10_000_000
+      self.updated[name] = True
+    self.messages["deviceState"] = NS(started=False, carBatteryCapacityUwh=1_000_000)
+    self.messages["pandaStates"] = [NS(ignitionLine=False, ignitionCan=False, pandaType="uno")]
+    self.messages["peripheralState"] = NS(pandaType="uno", voltage=12_500)
+    # A real Cap'n Proto event is serialized and decoded before projection.
+    event = messaging.new_message("accelerometer", valid=True)
+    event.accelerometer.timestamp = sample_stamp if sample_stamp is not None else now - 20_000_000
+    event.accelerometer.acceleration.v = list(acceleration)
+    return event.to_bytes()
+
+
+class TestRuntime(unittest.TestCase):
+  def setUp(self):
+    self.directory = tempfile.TemporaryDirectory()
+    self.addCleanup(self.directory.cleanup)
+    self.environment = patch.dict(os.environ, PARAMS_ROOT=self.directory.name, OPENPILOT_PREFIX="sentryruntime",
+                                  STARPILOT_SENTRY_DEVELOPMENT="1")
+    self.environment.start()
+    self.addCleanup(self.environment.stop)
+    self.params = Params()
+    self.params.put_bool("IsOffroad", True, block=True)
+    Path(self.params.get_param_path(KEY)).write_bytes(encode(Preferences(True)))
+    self.clock = Clock()
+    self.messages = Messages(self.clock)
+    self.events = []
+    self.runtime = SentryRuntime(self.params, self.messages, mono_clock=lambda: self.clock.mono,
+                                 boot_clock=self.clock.boot,
+                                 record=lambda kind, stamp, permitted: self.events.append((kind, stamp, permitted())))
+
+  def tick(self, *, stamp=None, acceleration=(0.0, 0.0, 9.81)):
+    self.clock.mono += 100_000_000
+    raw = self.messages.set_sources(sample_stamp=stamp, acceleration=acceleration)
+    with log.Event.from_bytes(raw) as parsed:
+      self.messages.messages["accelerometer"] = parsed.accelerometer
+      return self.runtime.tick()
+
+  def test_status_reports_actual_policy_and_immediate_sensor_revocation(self):
+    status = Mock()
+    self.runtime.status = status
+    decision = self.tick()
+    status.publish.assert_called_once_with(decision, encode(Preferences(True)), "idle")
+    self.assertEqual(decision.state, "arming")
+    self.messages.updated["accelerometer"] = False
+    self.clock.mono += 100_000_000
+    decision = self.runtime.tick()
+    self.assertEqual(decision.state, "sensor_unavailable")
+    self.assertEqual(status.publish.call_args.args[0], decision)
+
+  def test_serialized_sensor_and_fresh_parked_evidence(self):
+    first = self.tick()
+    self.assertEqual(first.state, "arming")
+    self.assertIsNotNone(self.runtime.policy.last_sample_time)
+    self.messages.updated["accelerometer"] = False
+    self.clock.mono += 100_000_000
+    self.assertEqual(self.runtime.tick().state, "sensor_unavailable")
+    self.assertIsNone(self.runtime.policy.arm_started)
+    self.assertEqual(self.events, [])
+
+
+  def test_serialized_device_panda_peripheral_use_actual_clock_domains(self):
+    self.clock.mono += 100_000_000
+    sensor_raw = self.messages.set_sources()
+    device = messaging.new_message("deviceState", valid=True)
+    device.deviceState.started = False
+    device.deviceState.carBatteryCapacityUwh = 1_000_000
+    panda = messaging.new_message("pandaStates", 1, valid=True)
+    panda.pandaStates[0].pandaType = "uno"
+    peripheral = messaging.new_message("peripheralState", valid=True)
+    peripheral.peripheralState.pandaType = "uno"
+    peripheral.peripheralState.voltage = 12_500
+    with ExitStack() as stack:
+      readers = [stack.enter_context(log.Event.from_bytes(raw)) for raw in
+                 (sensor_raw, device.to_bytes(), panda.to_bytes(), peripheral.to_bytes())]
+      self.messages.messages.update(accelerometer=readers[0].accelerometer, deviceState=readers[1].deviceState,
+                                    pandaStates=readers[2].pandaStates, peripheralState=readers[3].peripheralState)
+      self.assertEqual(self.runtime.tick().state, "arming")
+      # A C++ pandad BOOTTIME stamp incorrectly supplied in MONOTONIC units is stale.
+      self.messages.logMonoTime["pandaStates"] = self.clock.mono - 10_000_000
+      self.assertNotEqual(self.runtime.tick().state, "armed")
+
+  def test_unknown_ignition_power_and_shutdown_disarm(self):
+    self.tick()
+    self.messages.messages["pandaStates"][0].ignitionCan = True
+    self.assertEqual(self.runtime.tick().state, "disabled_ignition")
+    self.messages.messages["pandaStates"][0].ignitionCan = False
+    self.messages.messages["peripheralState"].voltage = 11_700
+    self.assertEqual(self.runtime.tick().state, "low_voltage")
+    self.messages.messages["peripheralState"].voltage = 12_500
+    self.params.put_bool("DoShutdown", True, block=True)
+    self.assertNotEqual(self.runtime.tick().state, "armed")
+    self.assertEqual(self.events, [])
+
+  def test_resume_barrier_rejects_queued_pre_resume_sensor(self):
+    self.tick()
+    old_stamp = self.messages.messages["accelerometer"].timestamp
+    self.clock.offset += 100_000_000
+    self.clock.mono += 100_000_000
+    raw = self.messages.set_sources(sample_stamp=old_stamp)
+    with log.Event.from_bytes(raw) as parsed:
+      self.messages.messages["accelerometer"] = parsed.accelerometer
+      self.assertNotEqual(self.runtime.tick().state, "armed")
+    self.assertIsNone(self.runtime.policy.arm_started)
+    self.assertEqual(self.events, [])
+    self.assertEqual(self.tick().state, "arming")
+
+  def test_revoked_settings_and_source_preserved(self):
+    self.assertEqual(self.tick().state, "arming")
+    raw = b'{"version":1,"enabled":true,"sensitivity":NaN,"warningTimeSeconds":1}'
+    Path(self.params.get_param_path(KEY)).write_bytes(raw)
+    self.assertEqual(self.tick().state, "disabled")
+    self.assertEqual(Path(self.params.get_param_path(KEY)).read_bytes(), raw)
+    self.assertEqual(self.events, [])
+
+  def test_full_arming_to_warning_records_once_with_fresh_final_authority(self):
+    for _ in range(901):
+      self.tick()
+    self.assertIsNotNone(self.runtime.policy.arm_started)
+    for index in range(11):
+      value = 10.0 if index % 2 else 9.81
+      self.tick(acceleration=(0.0, 0.0, value))
+    self.assertEqual(len(self.events), 1)
+    self.assertEqual(self.events[0][0], "warning")
+    self.assertTrue(self.events[0][2])
+    self.assertEqual(self.runtime.recording_state, "durability_unknown")  # Stub supplies no durability receipt.
+    self.params.put_bool("DoShutdown", True, block=True)
+    self.assertNotEqual(self.tick().state, "armed")
+    self.assertEqual(len(self.events), 1)
+
+  def test_store_callback_rechecks_saved_revision_and_sample_age(self):
+    self.tick()
+    self.runtime.policy.arm_started = self.clock.mono / 1e9 - 90
+    self.runtime.policy.last_sample_time = self.clock.mono / 1e9 - 0.02
+    original = Path(self.params.get_param_path(KEY)).read_bytes()
+    self.assertTrue(self.runtime._permitted(original))
+    Path(self.params.get_param_path(KEY)).write_bytes(encode(Preferences(True, self.runtime.policy.settings.__class__(0.05, 1.0))))
+    self.assertFalse(self.runtime._permitted(original))
+    Path(self.params.get_param_path(KEY)).write_bytes(original)
+    self.clock.mono += 600_000_000
+    self.assertFalse(self.runtime._permitted(original))
+
+
+class TestMainShutdown(unittest.TestCase):
+  def test_both_signals_close_owned_subscription_and_restore_handlers(self):
+    from openpilot.starpilot.sentry_mode import storage
+    for requested in (signal.SIGINT, signal.SIGTERM):
+      with self.subTest(requested=requested):
+        installed = {}
+        prior = object()
+        close = Mock()
+        def install(number, handler, target=installed):
+          target[number] = handler
+        def tick(which=requested, target=installed):
+          target[which](which, None)
+        with patch.object(signal, "getsignal", return_value=prior), patch.object(signal, "signal", side_effect=install), \
+             patch.object(messaging, "SubMaster"), patch.object(runtime_module, "Params"), \
+             patch.object(storage, "EventStore"), patch.object(runtime_module, "SentryRuntime") as owner:
+          owner.return_value.tick.side_effect = tick
+          owner.return_value.close = close
+          runtime_module.main()
+        close.assert_called_once()
+        self.assertIs(installed[signal.SIGINT], prior)
+        self.assertIs(installed[signal.SIGTERM], prior)
+
+
+if __name__ == "__main__":
+  unittest.main()

openpilot/starpilot/sentry_mode/tests/test_status.py

@@ -0,0 +1,68 @@
+import os
+from pathlib import Path
+import tempfile
+import unittest
+from unittest.mock import patch
+
+from openpilot.starpilot.sentry_mode.policy import Decision
+from openpilot.starpilot.sentry_mode.preferences import Preferences, SavedPreferences, encode
+from openpilot.starpilot.sentry_mode.status import RuntimeStatus, TTL_NS
+
+
+class RuntimeStatusTest(unittest.TestCase):
+  def setUp(self):
+    self.directory = tempfile.TemporaryDirectory()
+    self.addCleanup(self.directory.cleanup)
+    self.now = 10_000_000_000
+    self.boot_now = self.now + 50_000_000_000
+    self.identity = "current-boot"
+    self.status = RuntimeStatus(Path(self.directory.name).resolve(), clock=lambda: self.now,
+                                boot=lambda: self.identity, boot_clock=lambda: self.boot_now)
+    preferences = Preferences(True)
+    self.saved = SavedPreferences(encode(preferences), True, True, preferences)
+    self.environment = patch.dict(os.environ, STARPILOT_SENTRY_DEVELOPMENT="1")
+    self.environment.start()
+    self.addCleanup(self.environment.stop)
+
+  def test_saved_enable_is_not_armed_and_fresh_countdown_is_observed(self):
+    self.assertEqual(self.status.snapshot(self.saved)[0], "Waiting for monitor")
+    self.status.publish(Decision("arming", seconds_remaining=82), self.saved.raw, "idle")
+    self.assertEqual(self.status.snapshot(self.saved)[0], "Arming · 82s")
+    self.status.publish(Decision("armed"), self.saved.raw, "stored")
+    self.assertEqual(self.status.snapshot(self.saved)[0], "Monitoring motion")
+    self.status.publish(Decision("disabled_ignition"), self.saved.raw, "stored")
+    self.assertEqual(self.status.snapshot(self.saved)[0], "Not armed")
+
+  def test_expired_future_reboot_resume_or_changed_settings_never_show_armed(self):
+    for failure in ("expired", "future", "reboot", "resume", "settings"):
+      with self.subTest(failure=failure):
+        self.now = 10_000_000_000
+        self.boot_now = self.now + 50_000_000_000
+        self.identity = "current-boot"
+        preferences = Preferences(True)
+        self.saved = SavedPreferences(encode(preferences), True, True, preferences)
+        self.status.last_published = None
+        self.status.publish(Decision("armed"), self.saved.raw, "idle")
+        if failure == "expired":
+          self.now += TTL_NS + 1
+        elif failure == "future":
+          self.now -= 1
+        elif failure == "reboot":
+          self.identity = "another-boot"
+        elif failure == "resume":
+          self.boot_now += 300_000_000
+        else:
+          self.saved = SavedPreferences(self.saved.raw + b" ", True, True, self.saved.preferences)
+        self.assertEqual(self.status.snapshot(self.saved)[0], "Waiting for monitor")
+
+  def test_legacy_development_flag_does_not_gate_real_monitor_and_unsafe_status_is_rejected(self):
+    self.status.publish(Decision("armed"), self.saved.raw, "idle")
+    with patch.dict(os.environ, STARPILOT_SENTRY_DEVELOPMENT="0"):
+      self.assertEqual(self.status.snapshot(self.saved)[0], "Monitoring motion")
+    destination = Path(self.directory.name) / "status.json"
+    original = destination.read_bytes()
+    destination.unlink()
+    target = Path(self.directory.name) / "other"
+    target.write_bytes(original)
+    destination.symlink_to(target)
+    self.assertEqual(self.status.snapshot(self.saved)[0], "Waiting for monitor")

openpilot/starpilot/sentry_mode/tests/test_storage.py

@@ -0,0 +1,142 @@
+import json
+import os
+from pathlib import Path
+import tempfile
+import unittest
+from unittest.mock import patch
+
+from openpilot.starpilot.sentry_mode import storage
+
+
+class EventStoreTest(unittest.TestCase):
+  def setUp(self):
+    temporary = tempfile.TemporaryDirectory()
+    self.addCleanup(temporary.cleanup)
+    # macOS TemporaryDirectory may use /var, itself a system symlink. The
+    # storage contract intentionally never follows application path symlinks.
+    self.root = Path(temporary.name).resolve() / "events"
+    self.store = storage.EventStore(self.root)
+
+  def record(self, kind="warning", now=90_000_000_000):
+    return self.store.record(kind, now, permitted=lambda: True)
+
+  def test_round_trip_private_metadata_only_and_missing_is_empty(self):
+    self.assertEqual(self.store.snapshot()["events"], [])
+    first = self.record()
+    second = self.record("alarm", 120_000_000_000)
+    self.assertTrue(first.durable and second.durable)
+    events = self.store.snapshot()["events"]
+    self.assertEqual([event["kind"] for event in events], ["alarm", "warning"])
+    self.assertEqual({event["sessionId"] for event in events}, {self.store.session_id})
+    self.assertEqual(set(events[0]), storage.FIELDS)
+    self.assertEqual(self.root.stat().st_mode & 0o777, 0o700)
+    for path in self.root.iterdir():
+      self.assertEqual(path.stat().st_mode & 0o777, 0o600)
+      self.assertLessEqual(path.stat().st_size, storage.MAX_EVENT_BYTES)
+
+  def test_onroad_before_or_during_write_does_not_publish(self):
+    for decisions in ((False,), (True, False)):
+      with self.assertRaises(storage.StorageUnavailable):
+        self.store.record("warning", 100, permitted=iter(decisions).__next__)
+      self.assertEqual(self.store.snapshot()["events"], [])
+    self.assertFalse(any(path.name.endswith(".tmp") for path in self.root.iterdir()))
+
+  def test_capacity_stops_writes_without_deleting_existing_evidence(self):
+    with patch.object(storage, "MAX_EVENTS", 2):
+      self.record()
+      self.record("alarm")
+      before = {path.name: path.read_bytes() for path in self.root.iterdir()}
+      with self.assertRaisesRegex(storage.StorageUnavailable, "full"):
+        self.record()
+      self.assertEqual(before, {path.name: path.read_bytes() for path in self.root.iterdir()})
+      self.assertEqual(len(self.store.snapshot()["events"]), 2)
+
+  def test_refuses_symlink_directory_lock_or_event_without_touching_target(self):
+    target = self.root.parent / "unrelated"
+    target.mkdir(mode=0o700)
+    self.root.symlink_to(target)
+    with self.assertRaises(storage.StorageUnavailable):
+      self.record()
+    self.root.unlink()
+    self.root.mkdir(mode=0o700)
+    outside = target / "content"
+    outside.write_bytes(b"keep")
+    (self.root / ".lock").symlink_to(outside)
+    with self.assertRaises(storage.StorageUnavailable):
+      self.record()
+    (self.root / ".lock").unlink()
+    (self.root / ("a" * 32 + ".json")).symlink_to(outside)
+    snapshot = self.store.snapshot()
+    self.assertTrue(snapshot["incomplete"])
+    self.assertEqual(snapshot["events"], [])
+    self.assertEqual(outside.read_bytes(), b"keep")
+
+  def test_refuses_symlinked_parent_before_mkdir_or_publication(self):
+    outside = self.root.parent / "outside"
+    outside.mkdir(mode=0o700)
+    redirected = outside / "sentry"
+    redirected.mkdir(mode=0o700)
+    parent = self.root.parent / "alias"
+    parent.symlink_to(outside, target_is_directory=True)
+    store = storage.EventStore(parent / "sentry" / "events")
+    with self.assertRaises(storage.StorageUnavailable):
+      store.record("warning", 100, permitted=lambda: True)
+    with self.assertRaises(storage.StorageUnavailable):
+      store.snapshot()
+    self.assertFalse((redirected / "events").exists())
+    self.assertEqual(list(redirected.iterdir()), [])
+
+  def test_first_use_creates_only_private_directories_without_dotdot(self):
+    root = self.root.parent / "new" / "sentry" / "events"
+    store = storage.EventStore(root)
+    self.assertTrue(store.record("warning", 100, permitted=lambda: True).durable)
+    for parent in (root, root.parent, root.parent.parent):
+      self.assertEqual(parent.stat().st_mode & 0o777, 0o700)
+    escaped = storage.EventStore(self.root.parent / "new" / ".." / "outside" / "event")
+    with self.assertRaises(storage.StorageUnavailable):
+      escaped.record("warning", 100, permitted=lambda: True)
+
+  def test_invalid_duplicate_oversized_unknown_files_are_bounded_and_skipped(self):
+    first = self.record()
+    valid = self.root / (first.event_id + ".json")
+    value = json.loads(valid.read_bytes())
+    for index, raw in enumerate((b"{" + valid.read_bytes()[1:-1] + b',"version":1}',
+                                 b"x" * (storage.MAX_EVENT_BYTES + 1),
+                                 json.dumps({**value, "imagePaths": ["/private"]}).encode())):
+      path = self.root / (str(index) * 32 + ".json")
+      path.write_bytes(raw)
+      path.chmod(0o600)
+    (self.root / "unrecognized").write_bytes(b"keep")
+    result = self.store.snapshot()
+    self.assertTrue(result["incomplete"])
+    self.assertEqual(len(result["events"]), 1)
+    with patch.object(storage, "MAX_EVENTS", 1):
+      self.assertTrue(self.store.snapshot()["incomplete"])
+
+  def test_fsync_error_after_publication_reports_uncertain_durability_without_retry(self):
+    original = os.fsync
+    def fail_directory(fd):
+      if storage.stat.S_ISDIR(os.fstat(fd).st_mode):
+        raise OSError("disk sync failed")
+      original(fd)
+    with patch.object(storage.os, "fsync", side_effect=fail_directory):
+      receipt = self.record()
+    self.assertFalse(receipt.durable)
+    self.assertEqual([event["eventId"] for event in self.store.snapshot()["events"]], [receipt.event_id])
+
+  def test_fsync_error_before_publication_leaves_no_record(self):
+    with patch.object(storage.os, "fsync", side_effect=OSError("disk sync failed")):
+      with self.assertRaises(storage.StorageUnavailable):
+        self.record()
+    self.assertEqual(self.store.snapshot()["events"], [])
+    self.assertEqual([path.name for path in self.root.iterdir()], [".lock"])
+
+  def test_event_and_timestamps_are_strict(self):
+    for kind, stamp in (("image", 1), ("warning", True), ("warning", -1), ("alarm", 2**63)):
+      with self.assertRaises(ValueError):
+        self.record(kind, stamp)
+    self.assertFalse(self.root.exists())
+
+
+if __name__ == "__main__":
+  unittest.main()

openpilot/starpilot/sentry_mode/tests/test_storage_capacity.py

@@ -0,0 +1,29 @@
+from pathlib import Path
+import tempfile
+import unittest
+from unittest.mock import patch
+
+from openpilot.starpilot.sentry_mode import storage
+
+
+class TestStorageCapacity(unittest.TestCase):
+  def test_camera_events_fill_advertised_capacity_and_survive_restart(self):
+    with tempfile.TemporaryDirectory() as folder, patch.object(storage, "MAX_EVENTS", 4):
+      owner = storage.EventStore(Path(folder).resolve() / "events")
+      image = b"\xff\xd8frame\xff\xd9"
+      for stamp in range(4):
+        self.assertTrue(owner.record("warning", stamp, permitted=lambda: True, images={"wide": image, "cabin": image}).durable)
+      restarted = storage.EventStore(owner.root)
+      self.assertEqual(len(restarted.snapshot()["events"]), 4)
+      self.assertEqual(restarted.snapshot()["capacity"], 4)
+      with self.assertRaises(storage.StorageUnavailable):
+        restarted.record("alarm", 5, permitted=lambda: True, images={"cabin": image})
+
+  def test_unpublished_image_directory_still_consumes_capacity(self):
+    with tempfile.TemporaryDirectory() as folder, patch.object(storage, "MAX_EVENTS", 2):
+      owner = storage.EventStore(Path(folder).resolve() / "events")
+      owner.root.mkdir(mode=0o700)
+      (owner.root / ("a" * 32)).mkdir(mode=0o700)
+      owner.record("warning", 1, permitted=lambda: True)
+      with self.assertRaises(storage.StorageUnavailable):
+        owner.record("warning", 2, permitted=lambda: True)

openpilot/starpilot/sentry_mode/web_push.py

@@ -0,0 +1,85 @@
+"""RFC8291 aes128gcm and RFC8292 VAPID using the existing JWT crypto dependency."""
+import base64
+import json
+import os
+import struct
+import time
+from urllib.parse import urlsplit
+
+from cryptography.hazmat.primitives import hashes, serialization
+from cryptography.hazmat.primitives.asymmetric import ec, utils
+from cryptography.hazmat.primitives.ciphers.aead import AESGCM
+from cryptography.hazmat.primitives.kdf.hkdf import HKDF
+
+
+def b64(raw: bytes) -> str:
+  return base64.urlsafe_b64encode(raw).decode().rstrip('=')
+
+
+def unb64(value: str) -> bytes:
+  return base64.b64decode(value + '=' * (-len(value) % 4), altchars=b'-_', validate=True)
+
+
+def public(key) -> bytes:
+  return key.public_key().public_bytes(serialization.Encoding.X962, serialization.PublicFormat.UncompressedPoint)
+
+
+def new_key() -> str:
+  key = ec.generate_private_key(ec.SECP256R1())
+  return key.private_bytes(serialization.Encoding.PEM, serialization.PrivateFormat.PKCS8,
+                           serialization.NoEncryption()).decode()
+
+
+def load_key(pem: str):
+  key = serialization.load_pem_private_key(pem.encode(), password=None)
+  if not isinstance(key, ec.EllipticCurvePrivateKey) or not isinstance(key.curve, ec.SECP256R1):
+    raise ValueError('Invalid push key')
+  return key
+
+
+def validate_subscription(subscription: dict) -> None:
+  if type(subscription) is not dict or set(subscription) != {'endpoint', 'keys'}:
+    raise ValueError('Invalid push subscription')
+  keys = subscription['keys']
+  if type(keys) is not dict or set(keys) != {'p256dh', 'auth'}:
+    raise ValueError('Invalid push keys')
+  if any(type(value) is not str or len(value) > 256 for value in keys.values()):
+    raise ValueError('Invalid push keys')
+  receiver = unb64(keys['p256dh'])
+  if len(receiver) != 65 or len(unb64(keys['auth'])) != 16:
+    raise ValueError('Invalid push keys')
+  ec.EllipticCurvePublicKey.from_encoded_point(ec.SECP256R1(), receiver)
+
+
+def encrypt(subscription: dict, payload: bytes, *, sender=None, salt: bytes | None = None) -> bytes:
+  validate_subscription(subscription)
+  receiver = unb64(subscription['keys']['p256dh'])
+  auth = unb64(subscription['keys']['auth'])
+  sender = sender if sender is not None else ec.generate_private_key(ec.SECP256R1())
+  sender_public = public(sender)
+  shared = sender.exchange(ec.ECDH(), ec.EllipticCurvePublicKey.from_encoded_point(ec.SECP256R1(), receiver))
+  ikm = HKDF(algorithm=hashes.SHA256(), length=32, salt=auth,
+             info=b'WebPush: info\x00' + receiver + sender_public).derive(shared)
+  salt = salt if salt is not None else os.urandom(16)
+  cek = HKDF(algorithm=hashes.SHA256(), length=16, salt=salt, info=b'Content-Encoding: aes128gcm\x00').derive(ikm)
+  nonce = HKDF(algorithm=hashes.SHA256(), length=12, salt=salt, info=b'Content-Encoding: nonce\x00').derive(ikm)
+  plaintext = payload + b'\x02'
+  if len(plaintext) > 3000:
+    raise ValueError('Push payload exceeds limit')
+  return salt + struct.pack('!I', 4096) + bytes([65]) + sender_public + AESGCM(cek).encrypt(nonce, plaintext, None)
+
+
+def request(subscription: dict, message: dict, pem: str, *, now: int | None = None) -> tuple[bytes, dict]:
+  body = encrypt(subscription, json.dumps(message, separators=(',', ':')).encode())
+  parsed = urlsplit(subscription['endpoint'])
+  origin = f'{parsed.scheme}://{parsed.netloc}'
+  header = b64(b'{"typ":"JWT","alg":"ES256"}')
+  # RFC8292 expiration is a Unix epoch, independent of process monotonic time.
+  claims = b64(json.dumps({'aud': origin, 'exp': (time.time_ns() // 1_000_000_000 if now is None else now) + 43200,
+                          'sub': os.getenv('STARPILOT_VAPID_SUBJECT', 'mailto:galaxy@firestar.link')}, separators=(',', ':')).encode())
+  signing = f'{header}.{claims}'.encode()
+  vapid = load_key(pem)
+  r, s = utils.decode_dss_signature(vapid.sign(signing, ec.ECDSA(hashes.SHA256())))
+  token = signing.decode() + '.' + b64(r.to_bytes(32, 'big') + s.to_bytes(32, 'big'))
+  return body, {'Authorization': f'vapid t={token}, k={b64(public(vapid))}', 'Content-Encoding': 'aes128gcm',
+                'Content-Type': 'application/octet-stream', 'TTL': '300', 'Topic': message['eventId'][:32]}