diff --git a/openpilot/starpilot/galaxy/camera_request.py b/openpilot/starpilot/galaxy/camera_request.py new file mode 100644 index 0000000000..bf1dedc47a --- /dev/null +++ b/openpilot/starpilot/galaxy/camera_request.py @@ -0,0 +1,87 @@ +"""Short, namespaced camera leases. Manager alone owns camera processes.""" +import fcntl +import hashlib +import os +from pathlib import Path +import stat +import tempfile +import time + +TTL_NS = 15_000_000_000 +FRAME_TTL_NS = 5_000_000_000 + + +def boot_ns(): + return time.clock_gettime_ns(getattr(time, 'CLOCK_BOOTTIME', time.CLOCK_MONOTONIC)) + + +def _path(name): + prefix = os.environ.get('OPENPILOT_PREFIX', 'd') + namespace = hashlib.sha256(prefix.encode()).hexdigest()[:16] + return Path('/dev/shm') / f'starpilot-camera-{os.getuid()}-{namespace}' / name + + +def _read(path): + fd = os.open(path, os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK) + try: + info = os.fstat(fd) + if not stat.S_ISREG(info.st_mode) or info.st_uid != os.getuid() or info.st_size > 32: + raise ValueError('Invalid camera lease') + return int(os.read(fd, 32)) + finally: + os.close(fd) + + +def _write(path, value): + path = Path(path) + path.parent.mkdir(mode=0o700, parents=True, exist_ok=True) + info = path.parent.lstat() + if not stat.S_ISDIR(info.st_mode) or info.st_uid != os.getuid() or info.st_mode & 0o077: + raise OSError('Camera lease directory is not private') + lock = os.open(path.with_name(path.name + '.lock'), os.O_WRONLY | os.O_CREAT | os.O_NOFOLLOW | os.O_NONBLOCK, 0o600) + temporary = None + try: + info = os.fstat(lock) + if not stat.S_ISREG(info.st_mode) or info.st_uid != os.getuid(): + raise OSError('Invalid camera lease lock') + fcntl.flock(lock, fcntl.LOCK_EX) + try: + previous = _read(path) + if abs(previous - value) <= TTL_NS: + value = max(value, previous) + except (OSError, ValueError): + pass + fd, temporary = tempfile.mkstemp(prefix=path.name + '.', dir=path.parent) + try: + os.write(fd, str(value).encode()) + finally: + os.close(fd) + os.replace(temporary, path) + temporary = None + finally: + if temporary is not None: + os.unlink(temporary) + os.close(lock) + + +def request(clock=boot_ns, path=None): + _write(path or _path('lease'), clock() + TTL_NS) + + +def requested(clock=boot_ns, path=None): + try: + remaining = _read(path or _path('lease')) - clock() + return 0 < remaining <= TTL_NS + except (OSError, ValueError): + return False + + +def mark_frame(clock=boot_ns, path=None): + _write(path or _path('cabin-frame'), clock()) + + +def frame_available(clock=boot_ns, path=None): + try: + return 0 <= clock() - _read(path or _path('cabin-frame')) <= FRAME_TTL_NS + except (OSError, ValueError): + return False diff --git a/openpilot/starpilot/galaxy/camera_snapshot.py b/openpilot/starpilot/galaxy/camera_snapshot.py new file mode 100644 index 0000000000..b512c402e4 --- /dev/null +++ b/openpilot/starpilot/galaxy/camera_snapshot.py @@ -0,0 +1,71 @@ +"""One bounded parked snapshot from an already-running camera owner.""" + +import subprocess +import sys +import threading +import time + + +CAMERAS = frozenset(("cabin", "wide", "narrow")) +MAX_JPEG_BYTES = 1_000_000 +_CAPTURE_LOCK = threading.Lock() + + +class SnapshotUnavailable(Exception): + pass + + +class SnapshotDenied(SnapshotUnavailable): + pass + + +class CameraSnapshot: + def __init__(self, *, spawn=subprocess.Popen, clock=time.monotonic): + self.spawn, self.clock = spawn, clock + self.last_request = {} + + def capture(self, camera: str, *, permitted) -> bytes: + if camera not in CAMERAS: + raise ValueError("Unknown camera") + if not permitted(): + raise SnapshotDenied + if not _CAPTURE_LOCK.acquire(blocking=False): + raise SnapshotUnavailable("Another snapshot is in progress") + process = None + try: + if self.clock() - self.last_request.get(camera, float('-inf')) < 1: + raise SnapshotUnavailable("Wait before another snapshot") + self.last_request[camera] = self.clock() + from openpilot.starpilot.galaxy.camera_request import request, boot_ns + request() + frame_after = boot_ns() + deadline = self.clock() + 12 + process = self.spawn([sys.executable, '-m', 'openpilot.starpilot.galaxy.camera_snapshot_worker', camera], + stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, close_fds=True) + while self.clock() < deadline: + if not permitted(): + raise SnapshotDenied + try: + body, _ = process.communicate(timeout=min(.1, max(.001, deadline - self.clock()))) + except subprocess.TimeoutExpired: + continue + if not permitted(): + raise SnapshotDenied + if (process.returncode != 0 or not 4 <= len(body) <= MAX_JPEG_BYTES or + not body.startswith(b'\xff\xd8') or not body.endswith(b'\xff\xd9')): + raise SnapshotUnavailable("No fresh camera frame") + if camera == "cabin": + from openpilot.starpilot.galaxy.camera_request import mark_frame + mark_frame(clock=lambda: frame_after) + return body + raise SnapshotUnavailable("Camera timed out") + except OSError as error: + raise SnapshotUnavailable("Camera unavailable") from error + finally: + if process is not None: + if process.poll() is None: + process.kill() + process.wait() + if process.stdout is not None: + process.stdout.close() + _CAPTURE_LOCK.release() diff --git a/openpilot/starpilot/galaxy/camera_snapshot_worker.py b/openpilot/starpilot/galaxy/camera_snapshot_worker.py new file mode 100644 index 0000000000..9eebffefc8 --- /dev/null +++ b/openpilot/starpilot/galaxy/camera_snapshot_worker.py @@ -0,0 +1,61 @@ +"""CPU-only frame reader. Does not start or stop camerad or change Params.""" + +from io import BytesIO +import sys +import time + +from PIL import Image + +from msgq.visionipc import VisionIpcClient +from openpilot.cereal.visionipc import VisionStreamType +from openpilot.starpilot.galaxy.camera_snapshot import MAX_JPEG_BYTES +from openpilot.system.camerad.snapshot import extract_image + + +STREAMS = {"cabin": VisionStreamType.VISION_STREAM_CABIN, "wide": VisionStreamType.VISION_STREAM_WIDE_ROAD, + "narrow": VisionStreamType.VISION_STREAM_NARROW_ROAD} + + +def boot_ns(): + return time.clock_gettime_ns(getattr(time, "CLOCK_BOOTTIME", time.CLOCK_MONOTONIC)) + + +def capture(camera, *, client_factory=VisionIpcClient, clock=boot_ns): + started = clock() + client = client_factory("camerad", STREAMS[camera], True) + while not client.connect(False): + if clock() - started > 9_000_000_000: + return None + time.sleep(.1) + while clock() - started < 10_000_000_000: + frame = client.recv(timeout_ms=100) + now = clock() + # Native camerad does not populate VIPC extra.valid; use capture freshness. + if frame is None or not started < client.timestamp_sof <= client.timestamp_eof <= now: + continue + if now - client.timestamp_sof > 500_000_000: + continue + width, height, stride, uv_offset = frame.width, frame.height, frame.stride, frame.uv_offset + uv_height = ((height // 2 + 15) // 16) * 16 + if (not 0 < width <= 4096 or not 0 < height <= 4096 or width % 2 or height % 2 or + not width <= stride <= 8192 or stride % 2 or not stride * height <= uv_offset <= stride * 4096 or + uv_offset % stride or len(frame.data) < uv_offset + stride * uv_height): + return None + image = Image.fromarray(extract_image(frame)) + image.thumbnail((1280, 720)) + output = BytesIO() + image.save(output, format="JPEG", quality=85) + body = output.getvalue() + return body if len(body) <= MAX_JPEG_BYTES else None + return None + + +if __name__ == '__main__': + try: + result = capture(sys.argv[1]) if len(sys.argv) == 2 and sys.argv[1] in STREAMS else None + if result is not None: + sys.stdout.buffer.write(result) + else: + sys.exit(1) + except Exception: + sys.exit(1) diff --git a/openpilot/starpilot/galaxy/tests/test_camera_request.py b/openpilot/starpilot/galaxy/tests/test_camera_request.py new file mode 100644 index 0000000000..1a3db47525 --- /dev/null +++ b/openpilot/starpilot/galaxy/tests/test_camera_request.py @@ -0,0 +1,32 @@ +from concurrent.futures import ThreadPoolExecutor +import os +from pathlib import Path +import tempfile +import unittest +from unittest.mock import patch + +from openpilot.starpilot.galaxy import camera_request + + +class TestCameraRequest(unittest.TestCase): + def test_concurrent_latest_expiry_and_finite_lease(self): + with tempfile.TemporaryDirectory() as directory: + path = Path(directory) / 'lease' + with ThreadPoolExecutor(max_workers=8) as pool: + list(pool.map(lambda n: camera_request.request(clock=lambda: 100 + n, path=path), range(128))) + self.assertEqual(int(path.read_text()), 227 + camera_request.TTL_NS) + self.assertTrue(camera_request.requested(clock=lambda: 228, path=path)) + self.assertFalse(camera_request.requested(clock=lambda: 227 + camera_request.TTL_NS, path=path)) + self.assertFalse(camera_request.requested(clock=lambda: 1, path=path)) + + def test_prefix_isolation_and_frame_receipt_aging(self): + with patch.dict(os.environ, OPENPILOT_PREFIX='camera-a'): + first = camera_request._path('lease') + with patch.dict(os.environ, OPENPILOT_PREFIX='camera-b'): + self.assertNotEqual(first, camera_request._path('lease')) + with tempfile.TemporaryDirectory() as directory: + path = Path(directory) / 'frame' + camera_request.mark_frame(clock=lambda: 100, path=path) + self.assertTrue(camera_request.frame_available(clock=lambda: 101, path=path)) + self.assertFalse(camera_request.frame_available(clock=lambda: 101 + camera_request.FRAME_TTL_NS, path=path)) + self.assertFalse(camera_request.frame_available(clock=lambda: 99, path=path)) diff --git a/openpilot/starpilot/galaxy/tests/test_camera_snapshot.py b/openpilot/starpilot/galaxy/tests/test_camera_snapshot.py new file mode 100644 index 0000000000..4cc8078380 --- /dev/null +++ b/openpilot/starpilot/galaxy/tests/test_camera_snapshot.py @@ -0,0 +1,125 @@ +from io import BytesIO +import subprocess +import sys +from types import SimpleNamespace +import unittest +from unittest.mock import patch + +from PIL import Image +import numpy as np + +from openpilot.starpilot.galaxy.camera_snapshot import CameraSnapshot, SnapshotDenied, SnapshotUnavailable +from openpilot.starpilot.galaxy.camera_snapshot_worker import capture + + +class CameraSnapshotTest(unittest.TestCase): + def setUp(self): + for name in ("request", "mark_frame"): + replacement = patch("openpilot.starpilot.galaxy.camera_request." + name) + setattr(self, name + "_mock", replacement.start()) + self.addCleanup(replacement.stop) + + def test_denied_does_not_launch_reader(self): + owner = CameraSnapshot(spawn=lambda *args, **kwargs: self.fail("launched")) + with self.assertRaises(SnapshotDenied): + owner.capture('cabin', permitted=lambda: False) + with self.assertRaises(ValueError): + owner.capture('../camera', permitted=lambda: True) + + def test_revocation_kills_and_reaps_blocked_reader(self): + children = [] + def spawn(args, **kwargs): + self.assertEqual(args[-1], 'cabin') + with self.assertRaises(SnapshotUnavailable): + CameraSnapshot(spawn=lambda *a, **k: self.fail("concurrent reader")).capture( + 'wide', permitted=lambda: True) + children.append(subprocess.Popen([sys.executable, '-c', 'import time; time.sleep(30)'], **kwargs)) + return children[-1] + calls = 0 + def permitted(): + nonlocal calls + calls += 1 + return calls < 3 + with self.assertRaises(SnapshotDenied): + CameraSnapshot(spawn=spawn).capture('cabin', permitted=permitted) + self.assertIsNotNone(children[0].poll()) + self.assertTrue(children[0].stdout.closed) + + def test_deadline_kills_blocked_native_reader(self): + children = [] + ticks = iter(range(100)) + def spawn(args, **kwargs): + children.append(subprocess.Popen([sys.executable, '-c', 'import time; time.sleep(30)'], **kwargs)) + return children[-1] + with self.assertRaises(SnapshotUnavailable): + CameraSnapshot(spawn=spawn, clock=lambda: next(ticks)).capture('cabin', permitted=lambda: True) + self.assertIsNotNone(children[0].poll()) + self.assertTrue(children[0].stdout.closed) + + def test_reader_result_and_final_authority_check(self): + def spawn(args, **kwargs): + return subprocess.Popen([sys.executable, '-c', "import sys; sys.stdout.buffer.write(bytes([255,216,255,217]))"], **kwargs) + owner = CameraSnapshot(spawn=spawn) + self.assertEqual(owner.capture('wide', permitted=lambda: True), b'\xff\xd8\xff\xd9') + self.assertEqual(owner.capture('cabin', permitted=lambda: True), b'\xff\xd8\xff\xd9') + self.assertEqual(self.request_mock.call_count, 2) + self.assertEqual(self.mark_frame_mock.call_count, 1) # Cabin availability receipt only. + with self.assertRaises(SnapshotUnavailable): + owner.capture('wide', permitted=lambda: True) + calls = 0 + def permitted(): + nonlocal calls + calls += 1 + return calls < 3 + with self.assertRaises(SnapshotDenied): + CameraSnapshot(spawn=spawn).capture('wide', permitted=permitted) + + def test_real_nv12_conversion_accepts_only_post_request_fresh_native_frame(self): + now = 1_000_000_000 + data = np.full(80, 128, dtype=np.uint8) + frame = SimpleNamespace(width=4, height=4, stride=4, uv_offset=16, data=data) + class Client: + valid = False # Native camerad leaves the extra metadata bit unset. + timestamp_sof = now - 1 + timestamp_eof = now - 1 + frames = 0 + def connect(self, blocking): + self.blocking = blocking + return True + def recv(self, timeout_ms): + nonlocal now + self.frames += 1 + now += 100_000_000 + if self.frames > 1: + self.timestamp_sof = now - 20_000_000 + self.timestamp_eof = now - 10_000_000 + return frame + client = Client() + result = capture('cabin', client_factory=lambda *args: client, clock=lambda: now) + self.assertFalse(client.blocking) + self.assertEqual(client.frames, 2) + image = Image.open(BytesIO(result)) + self.assertEqual(image.size, (4, 4)) + self.assertEqual(image.mode, 'RGB') + frame.width = 99999 + self.assertIsNone(capture('cabin', client_factory=lambda *args: client, clock=lambda: now)) + + + def test_native_frames_with_stale_future_or_reversed_timestamps_never_convert(self): + for fault in ('stale', 'future', 'reversed'): + with self.subTest(fault=fault): + now = [1_000_000_000] + frame = SimpleNamespace(width=4, height=4, stride=4, uv_offset=16, + data=np.full(80, 128, dtype=np.uint8)) + class Client: + valid = False + timestamp_sof = timestamp_eof = 0 + def connect(self, blocking): return True + def recv(self, timeout_ms, *, fault=fault, frame=frame, now=now): + now[0] += 100_000_000 + self.timestamp_sof = now[0] - (600_000_000 if fault == 'stale' else 20_000_000) + self.timestamp_eof = now[0] + 1 if fault == 'future' else self.timestamp_sof - 1 if fault == 'reversed' else now[0] - 10_000_000 + return frame + with patch('openpilot.starpilot.galaxy.camera_snapshot_worker.extract_image') as convert: + self.assertIsNone(capture('cabin', client_factory=lambda *args: Client(), clock=lambda now=now: now[0])) + convert.assert_not_called() diff --git a/openpilot/starpilot/galaxy/tests/test_cameras.mjs b/openpilot/starpilot/galaxy/tests/test_cameras.mjs new file mode 100644 index 0000000000..f0f8d288d7 --- /dev/null +++ b/openpilot/starpilot/galaxy/tests/test_cameras.mjs @@ -0,0 +1,66 @@ +import assert from "node:assert/strict" +import { readFileSync } from "node:fs" +import { CamerasPage, CameraSnapshotFeed } from "../web/js/cameras.js" +import { SettingsFeed, SettingsPage } from "../web/js/settings.js" + +const requests = [] +const states = [] +const feed = new SettingsFeed({ publish: (state) => states.push(state), + fetcher: (url, options) => new Promise((resolve) => requests.push({ url, options, resolve })), + later: () => 1, cancelTimer: () => {} }) +feed.start("pip") +assert.equal(requests[0].url, "./api/settings/pages/pip") +assert.equal(feed.page, "pip") +feed.stop() +requests[0].resolve({ ok: true, status: 200, json: async () => ({ page: "pip", rows: [] }) }) +await Promise.resolve() +await Promise.resolve() +assert.equal(states.at(-1).status, "idle") // Hidden or signed-out page cannot restore saved values. + +const mounted = { mode: "local", initialPage: "pip", state: { developerOpen: false }, feed: { start(page) { assert.equal(page, "pip") } } } +SettingsPage.mounted.call(mounted) +let returned = 0 +const page = { initialPage: "pip", returnTo: () => returned++, state: { query: "anything", data: { page: "pip" } }, + feed: { load() { throw new Error("Root PiP Back must return to Cameras") } } } +SettingsPage.methods.back.call(page) +assert.equal(returned, 1) +assert.equal(page.state.query, "") + +assert.match(CamerasPage.template, /Adjust the camera crop with a live cabin preview/) +assert.doesNotMatch(CamerasPage.template, /live camera preview is unavailable/) +assert.match(CamerasPage.template, /Sentry[\s\S]*View motion events/) +assert.match(CamerasPage.template, /Saved motion settings/) +assert.match(CamerasPage.template, /V-ASM[\s\S]*Open saved settings/) +assert.match(CamerasPage.template, /live camera image and current warning status are unavailable/) +assert.match(CamerasPage.template, /snapshots.capture\(camera\)/) +assert.match(SettingsPage.template, /\{\{ title \}\}/) +const app = readFileSync(new URL("../web/js/app.js", import.meta.url), "utf8") +assert.match(app, / snapshots.push(value), unauthorized: () => snapshots.push("unauthorized"), + fetcher: (url, options) => new Promise((resolve) => captures.push({ url, options, resolve })), + createURL: () => "blob:camera", revokeURL: (url) => revoked.push(url), + later: (fn) => { timer = fn; return 1 }, cancelTimer: () => {} }) +const pending = camera.capture("cabin") +assert.equal(captures[0].url, "./api/cameras/snapshot") +assert.deepEqual(JSON.parse(captures[0].options.body), { camera: "cabin" }) +captures[0].resolve({ ok: true, status: 200, headers: new Map([["Content-Type", "image/jpeg; charset=binary"]]), blob: async () => ({ size: 100 }) }) +await pending +assert.equal(snapshots.at(-1).image, "blob:camera") +camera.stop() +assert.deepEqual(revoked, ["blob:camera"]) +const late = camera.capture("wide") +camera.stop() +assert.equal(captures[1].options.signal.aborted, true) +captures[1].resolve({ ok: true, status: 200, headers: new Map([["Content-Type", "image/jpeg"]]), blob: async () => ({ size: 100 }) }) +await late +assert.equal(snapshots.at(-1).image, "") +const timed = camera.capture("cabin") +timer() +captures[2].resolve({ status: 401 }) +await timed +assert.match(snapshots.at(-1).error, /timed out/)