"""MROI-01: one policy writer, CAS persistence, explicit desired/effective boundary.""" from contextlib import contextmanager from copy import deepcopy import fcntl import hashlib import json import os from pathlib import Path import stat import tempfile import threading from app.multi_roi.errors import RoiError from app.multi_roi.model import canonical, Snapshot DEFAULT = dict(mode='legacy', max_rois=8, max_rules=48, full_frame_roi_id=0) def validate(value): if type(value) is not dict or set(value) != set(DEFAULT): raise RoiError('INVALID_POLICY') if value['mode'] not in ('legacy', 'multi_roi'): raise RoiError('INVALID_POLICY', field='mode') # Parser/metadata bounds, not a claim that these sizes meet real-time capacity. for key, low, high in (('max_rois', 1, 256), ('max_rules', 1, 256), ('full_frame_roi_id', 0, 2**53-1)): if type(value[key]) is not int or not low <= value[key] <= high: raise RoiError('INVALID_POLICY', field=key) return dict(value) def digest(value): return hashlib.sha256(canonical(value).encode()).hexdigest() def safe_path(path): p = Path(path) if not p.is_absolute() or '..' in p.parts or any(q.is_symlink() for q in (p, *p.parents)): raise RoiError('INVALID_POLICY', field='path') repository = Path(__file__).resolve().parents[2] if p.is_relative_to(repository): raise RoiError('INVALID_POLICY', field='path') return p def read(path): fd = os.open(safe_path(path), os.O_RDONLY | os.O_NOFOLLOW) with os.fdopen(fd, 'rb') as handle: info = os.fstat(handle.fileno()) if not stat.S_ISREG(info.st_mode) or info.st_size > 16384: raise RoiError('INVALID_POLICY') raw = handle.read(16385) from app.secure_control import decode, ControlError try: return decode(raw) except ControlError: raise RoiError('INVALID_POLICY') from None def atomic(path, value): path = safe_path(path) previous = path.stat() if path.exists() else None fd, temporary = tempfile.mkstemp(prefix='.'+path.name+'.', dir=path.parent) try: with os.fdopen(fd, 'w', encoding='utf-8') as out: if previous is not None: os.fchmod(out.fileno(), stat.S_IMODE(previous.st_mode) & 0o666) if os.geteuid() == 0: os.fchown(out.fileno(), previous.st_uid, previous.st_gid) out.write(canonical(value)+'\n') out.flush() os.fsync(out.fileno()) os.replace(temporary, path) directory = os.open(path.parent, os.O_DIRECTORY) try: os.fsync(directory) finally: os.close(directory) finally: if os.path.exists(temporary): os.unlink(temporary) class PolicyStore: def __init__(self, path=None, *, command_lock=None, snapshots=lambda: ()): self.path = safe_path(path) if path is not None else None self.applied_path = self.path.with_name(self.path.stem+'.applied.json') if self.path else None self.lock = command_lock or threading.RLock() self.snapshots = snapshots self._value = dict(DEFAULT) if self.applied_path and self.applied_path.exists(): self._value = validate(read(self.applied_path)) # An edited desired file is never activated implicitly, including at boot. @property def value(self): # _value is replaced, never mutated. Pre-validation must not wait for a # Manager operation on the ASGI thread; commit revalidates under lock. return dict(self._value) def status(self): with self.lock: configured, error = None, None try: configured = validate(read(self.path)) if self.path and self.path.exists() else dict(DEFAULT) except (OSError, ValueError): error = 'INVALID_POLICY' return dict(path=str(self.path) if self.path else None, appliedPath=str(self.applied_path) if self.path else None, configured=configured, effective=dict(self._value), revision=digest(self._value), configuredRevision=digest(configured) if configured else None, pending=configured != self._value, error=error, nextAction='ct roi edit / ct roi apply') def check(self, candidate): for value in self.snapshots(): snapshot = Snapshot.restore(value) for kind, actual in (('rois', len(snapshot.areas)), ('rules', len(snapshot.rules))): if actual > candidate['max_'+kind]: raise RoiError('POLICY_LIMIT_CONFLICT', camera=snapshot.camera_id, field='max_'+kind, actual=actual, limit=candidate['max_'+kind]) if candidate['full_frame_roi_id'] != self._value['full_frame_roi_id']: if any(a.roi_id == candidate['full_frame_roi_id'] and not a.reserved for a in snapshot.areas): raise RoiError('POLICY_ID_CONFLICT', camera=snapshot.camera_id, roi=candidate['full_frame_roi_id']) @contextmanager def file_lock(self): if self.path is None: raise RoiError('POLICY_NOT_CONFIGURED') target = safe_path(self.path.with_suffix('.lock')) fd = os.open(target, os.O_CREAT | os.O_RDWR | os.O_NOFOLLOW, 0o600) try: fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB) yield except BlockingIOError: raise RoiError('POLICY_REVISION_CONFLICT') from None finally: os.close(fd) def apply(self, expected_revision, changes=None, *, configured_revision=None): with self.lock, self.file_lock(): disk = validate(read(self.applied_path)) if self.applied_path.exists() else dict(DEFAULT) if expected_revision != digest(self._value) or disk != self._value: raise RoiError('POLICY_REVISION_CONFLICT') before = validate(read(self.path)) if self.path.exists() else dict(DEFAULT) if configured_revision is None or configured_revision != digest(before): raise RoiError('POLICY_REVISION_CONFLICT') if changes is not None and (type(changes) is not dict or set(changes)-set(DEFAULT)): raise RoiError('INVALID_POLICY') candidate = validate(dict(before, **changes)) if changes is not None else before self.check(candidate) try: # Desired can remain ahead of effective after a failed write. Status exposes it; # only the atomic applied file defines restart behavior and current admission. if changes is not None or not self.path.exists(): atomic(self.path, candidate) atomic(self.applied_path, candidate) except OSError: raise RoiError('POLICY_PERSISTENCE_FAILED') from None self._value = deepcopy(candidate) return self.status()