From 2dd715a3d55ef7f9e483b9351deb9af2079ec289 Mon Sep 17 00:00:00 2001 From: firestar5683 <168790843+firestar5683@users.noreply.github.com> Date: Fri, 25 Sep 2026 14:53:20 -0500 Subject: [PATCH] Sentry Implement Sentry scanning, notification state and worker cancellation through explicit policies and preferences. --- openpilot/starpilot/sentry_mode/README.md | 49 ++ openpilot/starpilot/sentry_mode/__init__.py | 1 + openpilot/starpilot/sentry_mode/actions.py | 14 + .../starpilot/sentry_mode/notifications.py | 451 ++++++++++++++++++ openpilot/starpilot/sentry_mode/policy.py | 163 +++++++ .../starpilot/sentry_mode/preferences.py | 82 ++++ openpilot/starpilot/sentry_mode/runtime.py | 215 +++++++++ openpilot/starpilot/sentry_mode/status.py | 113 +++++ openpilot/starpilot/sentry_mode/storage.py | 285 +++++++++++ .../starpilot/sentry_mode/tests/__init__.py | 0 .../sentry_mode/tests/test_manager.py | 37 ++ .../sentry_mode/tests/test_notifications.py | 214 +++++++++ .../sentry_mode/tests/test_policy.py | 140 ++++++ .../sentry_mode/tests/test_preferences.py | 52 ++ .../sentry_mode/tests/test_runtime.py | 215 +++++++++ .../sentry_mode/tests/test_status.py | 68 +++ .../sentry_mode/tests/test_storage.py | 142 ++++++ .../tests/test_storage_capacity.py | 29 ++ openpilot/starpilot/sentry_mode/web_push.py | 85 ++++ 19 files changed, 2355 insertions(+) create mode 100644 openpilot/starpilot/sentry_mode/README.md create mode 100644 openpilot/starpilot/sentry_mode/__init__.py create mode 100644 openpilot/starpilot/sentry_mode/actions.py create mode 100644 openpilot/starpilot/sentry_mode/notifications.py create mode 100644 openpilot/starpilot/sentry_mode/policy.py create mode 100644 openpilot/starpilot/sentry_mode/preferences.py create mode 100644 openpilot/starpilot/sentry_mode/runtime.py create mode 100644 openpilot/starpilot/sentry_mode/status.py create mode 100644 openpilot/starpilot/sentry_mode/storage.py create mode 100644 openpilot/starpilot/sentry_mode/tests/__init__.py create mode 100644 openpilot/starpilot/sentry_mode/tests/test_manager.py create mode 100644 openpilot/starpilot/sentry_mode/tests/test_notifications.py create mode 100644 openpilot/starpilot/sentry_mode/tests/test_policy.py create mode 100644 openpilot/starpilot/sentry_mode/tests/test_preferences.py create mode 100644 openpilot/starpilot/sentry_mode/tests/test_runtime.py create mode 100644 openpilot/starpilot/sentry_mode/tests/test_status.py create mode 100644 openpilot/starpilot/sentry_mode/tests/test_storage.py create mode 100644 openpilot/starpilot/sentry_mode/tests/test_storage_capacity.py create mode 100644 openpilot/starpilot/sentry_mode/web_push.py diff --git a/openpilot/starpilot/sentry_mode/README.md b/openpilot/starpilot/sentry_mode/README.md new file mode 100644 index 0000000000..487e715353 --- /dev/null +++ b/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. diff --git a/openpilot/starpilot/sentry_mode/__init__.py b/openpilot/starpilot/sentry_mode/__init__.py new file mode 100644 index 0000000000..f71be47de9 --- /dev/null +++ b/openpilot/starpilot/sentry_mode/__init__.py @@ -0,0 +1 @@ +"""Parked Sentry motion policy, guarded observation and local event metadata.""" diff --git a/openpilot/starpilot/sentry_mode/actions.py b/openpilot/starpilot/sentry_mode/actions.py new file mode 100644 index 0000000000..6117c62e23 --- /dev/null +++ b/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_") diff --git a/openpilot/starpilot/sentry_mode/notifications.py b/openpilot/starpilot/sentry_mode/notifications.py new file mode 100644 index 0000000000..7102858753 --- /dev/null +++ b/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 diff --git a/openpilot/starpilot/sentry_mode/policy.py b/openpilot/starpilot/sentry_mode/policy.py new file mode 100644 index 0000000000..cba222e623 --- /dev/null +++ b/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") diff --git a/openpilot/starpilot/sentry_mode/preferences.py b/openpilot/starpilot/sentry_mode/preferences.py new file mode 100644 index 0000000000..fdeb50ee73 --- /dev/null +++ b/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 diff --git a/openpilot/starpilot/sentry_mode/runtime.py b/openpilot/starpilot/sentry_mode/runtime.py new file mode 100644 index 0000000000..24a266dc68 --- /dev/null +++ b/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() diff --git a/openpilot/starpilot/sentry_mode/status.py b/openpilot/starpilot/sentry_mode/status.py new file mode 100644 index 0000000000..a2c3be1864 --- /dev/null +++ b/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" diff --git a/openpilot/starpilot/sentry_mode/storage.py b/openpilot/starpilot/sentry_mode/storage.py new file mode 100644 index 0000000000..f3fbb903c8 --- /dev/null +++ b/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 [] diff --git a/openpilot/starpilot/sentry_mode/tests/__init__.py b/openpilot/starpilot/sentry_mode/tests/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/openpilot/starpilot/sentry_mode/tests/test_manager.py b/openpilot/starpilot/sentry_mode/tests/test_manager.py new file mode 100644 index 0000000000..78a92bdf5d --- /dev/null +++ b/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() diff --git a/openpilot/starpilot/sentry_mode/tests/test_notifications.py b/openpilot/starpilot/sentry_mode/tests/test_notifications.py new file mode 100644 index 0000000000..5318b3e53c --- /dev/null +++ b/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() diff --git a/openpilot/starpilot/sentry_mode/tests/test_policy.py b/openpilot/starpilot/sentry_mode/tests/test_policy.py new file mode 100644 index 0000000000..a689ccedab --- /dev/null +++ b/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) diff --git a/openpilot/starpilot/sentry_mode/tests/test_preferences.py b/openpilot/starpilot/sentry_mode/tests/test_preferences.py new file mode 100644 index 0000000000..35b0b01b69 --- /dev/null +++ b/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() diff --git a/openpilot/starpilot/sentry_mode/tests/test_runtime.py b/openpilot/starpilot/sentry_mode/tests/test_runtime.py new file mode 100644 index 0000000000..a3995ce2ba --- /dev/null +++ b/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() diff --git a/openpilot/starpilot/sentry_mode/tests/test_status.py b/openpilot/starpilot/sentry_mode/tests/test_status.py new file mode 100644 index 0000000000..7d58477b7e --- /dev/null +++ b/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") diff --git a/openpilot/starpilot/sentry_mode/tests/test_storage.py b/openpilot/starpilot/sentry_mode/tests/test_storage.py new file mode 100644 index 0000000000..0106c30f32 --- /dev/null +++ b/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() diff --git a/openpilot/starpilot/sentry_mode/tests/test_storage_capacity.py b/openpilot/starpilot/sentry_mode/tests/test_storage_capacity.py new file mode 100644 index 0000000000..cb0ca948e0 --- /dev/null +++ b/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) diff --git a/openpilot/starpilot/sentry_mode/web_push.py b/openpilot/starpilot/sentry_mode/web_push.py new file mode 100644 index 0000000000..c93219b32f --- /dev/null +++ b/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]}