"""Opt-in Runtime bridge: one existing CameraCommands owner and existing Manager lifecycle. This module intentionally does not import the Sidecar, GPU or a database. The Manager's command lock also fences legacy CLI/API writers during projection. """ import asyncio from contextlib import asynccontextmanager from copy import deepcopy from dataclasses import dataclass, field import hmac import json import logging import math from urllib.parse import urlsplit from uuid import uuid5, NAMESPACE_URL from fastapi import APIRouter, Request from fastapi.responses import JSONResponse CODES = frozenset({'FIRE','FALL','INTRUSION','LOITERING','FIGHT','CROWD'}) MAX_BODY = 65536 LOG = logging.getLogger(__name__) def command_error(error): import re value = str(error) if isinstance(error, ControlError) and re.fullmatch(r'[A-Z][A-Z0-9_]{0,63}', value): return value return 'RUNTIME_COMMAND_REJECTED' class ControlError(ValueError): pass def decode(raw): def pairs(items): d = {} for k, v in items: if k in d: raise ControlError('DUPLICATE_KEY') d[k] = v return d def bad(_): raise ControlError('INVALID_NUMBER') if len(raw) > MAX_BODY: raise ControlError('BODY_TOO_LARGE') try: d = json.loads(raw, object_pairs_hook=pairs, parse_constant=bad) if not isinstance(d, dict): raise ControlError('BODY_NOT_OBJECT') return d except (ValueError, UnicodeError, RecursionError): raise ControlError('INVALID_BODY') from None def camera_id(d): v = d.get('cameraId') if type(v) is not int or not 1 <= v <= 2**53-1: raise ControlError('CAMERA_ID_INVALID') return v def validate_start(d): camera_id(d) required = {'cameraId','rtspUrl','startWallTimeMs','rtpClockRate','scenarios'} if not required <= set(d) or set(d)-required-{'recServerId','recCameraId'}: raise ControlError('START_FIELDS_INVALID') u = d['rtspUrl'] try: p = urlsplit(u) if (not isinstance(u, str) or len(u)>4096 or p.scheme!='rtsp' or not p.hostname or p.port == 0 or p.fragment or any(c.isspace() or ord(c)<32 for c in u)): raise ValueError() except (TypeError, ValueError): raise ControlError('RTSP_INVALID') from None if type(d['startWallTimeMs']) is not int or not 0 < d['startWallTimeMs'] <= 2**53-1: raise ControlError('ANCHOR_INVALID') if type(d['rtpClockRate']) is not int or d['rtpClockRate'] != 90000: raise ControlError('CLOCK_RATE_UNSUPPORTED') scenarios = d['scenarios'] if not isinstance(scenarios, list) or len(scenarios) != 6: raise ControlError('SCENARIO_SNAPSHOT_REQUIRED') seen = set() for s in scenarios: if not isinstance(s, dict) or set(s)-{'rois'} != {'scenarioCode','enabled','confidence'}: raise ControlError('SCENARIO_FIELDS_INVALID') name = s['scenarioCode'] if not isinstance(name, str) or name not in CODES or name in seen or type(s['enabled']) is not bool: raise ControlError('SCENARIO_CODE_INVALID') seen.add(name) score = s['confidence'] if score is not None and (type(score) not in (int,float) or not math.isfinite(score) or not 0 <= score <= 1): raise ControlError('CONFIDENCE_INVALID') rois = s.get('rois') # Normalize before command canonicalization so equivalent empty ROI requests deduplicate. if rois is None: s['rois'] = rois = [] if not isinstance(rois, list) or len(rois) > 32: raise ControlError('ROI_LIST_INVALID') for roi in rois: if not isinstance(roi, dict) or set(roi) != {'label','points'}: raise ControlError('ROI_FIELDS_INVALID') label, points = roi['label'], roi['points'] if (not isinstance(label,str) or not label.strip() or len(label)>64 or any(ord(c)<32 for c in label) or not isinstance(points,list) or not 3<=len(points)<=32): raise ControlError('ROI_INVALID') for pair in points: if not isinstance(pair, list) or len(pair)!=2 or any(type(v) not in (float,int) or not math.isfinite(v) or not 0<=v<=1 for v in pair): raise ControlError('ROI_COORDINATES_INVALID') for k in ('recServerId','recCameraId'): if d.get(k) is not None and (not isinstance(d[k], str) or not d[k] or len(d[k])>256): raise ControlError('REC_ID_INVALID') # CROWD predev must never silently run the ordinary person provider. from app.crowd_runtime import validate_web_request try: validate_web_request(scenarios) except ValueError as exc: raise ControlError(str(exc)) from None @dataclass(frozen=True) class Start: # Canonical text makes duplicate in-flight requests compare by value, not object identity. text: str = field(repr=False) class ControlService: def __init__(self, manager, server_id, template=None): self.manager, self.server_id, self.template = manager, server_id, deepcopy(template) def start_camera(self, request): d = decode(request.text) validate_start(d) m = self.manager with m._commands: try: sid = m.camera_stream_id(str(d['cameraId'])) g = m.get(sid) except KeyError: g = None sid = 'camera-' + uuid5(NAMESPACE_URL, str(d['cameraId'])).hex if g is None: if self.template is None: raise ControlError('CAMERA_NOT_PROVISIONED') spec = deepcopy(self.template) spec.update(stream_id=sid, source_id=str(d['cameraId']), server_id=self.server_id, camera_id=d['cameraId'], source_type='rtsp', autostart=False) else: if g.spec.get('server_id') != self.server_id: raise ControlError('SERVER_ID_CONFLICT') # Never revive a Stop still cleaning up. Restart already owned by Manager is different. if g.observed_state == 'stopping' and g._pending_restart is None: raise ControlError('STOP_CLEANUP_PENDING') spec = deepcopy(m._saved_spec(g)) if spec.get('source_type', 'rtsp') != 'rtsp': raise ControlError('SOURCE_TYPE_CONFLICT') if spec.get('camera_id', d['cameraId']) != d['cameraId']: raise ControlError('CAMERA_ID_TYPE_CONFLICT') spec['uri'] = d['rtspUrl'] spec['camera_id'] = d['cameraId'] old_anchor = spec.get('frame_time_anchor') or {} # Replaying the same test Anchor must not promote it to a real clock source. spec['frame_time_anchor'] = {'startWallTimeMs': d['startWallTimeMs'], 'synthetic': bool(old_anchor.get('synthetic', False)) if old_anchor.get('startWallTimeMs') == d['startWallTimeMs'] else False} spec['tracking'] = dict(spec.get('tracking') or {}, enabled=True) intrusion = next(s for s in d['scenarios'] if s['scenarioCode']=='INTRUSION') cfg = dict(spec.get('intrusion') or {}) cfg.update(enabled=intrusion['enabled'], confidence=.5 if intrusion['confidence'] is None else intrusion['confidence']) roi = intrusion['rois'][0] if intrusion['rois'] else None cfg['roi'] = roi['points'] if roi else [[0.,0.],[1.,0.],[1.,1.],[0.,1.]] if roi: cfg['roi_name'] = roi['label'] else: cfg.pop('roi_name', None) spec['intrusion'] = cfg from app.region_scenario import project_loitering, validate as validate_region from app.fire_config import project_fire from app.fall_runtime import project_fall from app.fight_runtime import project_fight from app.crowd_runtime import project_crowd_candidate spec['scenarios'] = 'off' try: project_loitering(spec, d['scenarios']) project_fall(spec, d['scenarios']) project_fight(spec, d['scenarios']) project_fire(spec, d['scenarios']) spec = project_crowd_candidate(spec, d['scenarios']) validate_region(spec) except ValueError as exc: if str(exc) == 'NEEDS_MULTI_PROVIDER_RUNTIME_CONFIRMATION' or str(exc).startswith('CROWD_'): raise ControlError(str(exc)) from None raise ControlError('LOITERING_CONFIGURATION_UNSUPPORTED') from exc # Non-secret provenance: unsupported requests are explicit skips, not silent success. spec['secure_control_projection'] = { 'profile': 'REST_STATELESS', 'skippedScenarios': [s['scenarioCode'] for s in d['scenarios'] if s['enabled'] and s['scenarioCode'] not in ('INTRUSION', 'LOITERING', 'FIRE', 'FALL', 'FIGHT', 'CROWD')], 'ignoredIntrusionRois': max(0, len(intrusion['rois'])-1), 'requestedRtpClockRate': d['rtpClockRate'], } # Inference/model/source options and local storage knobs remain the saved values. spec['scenarios'] = 'off' for key in ('recServerId','recCameraId'): if key in d: spec['rec_server_id' if key=='recServerId' else 'rec_camera_id'] = d[key] return m.apply_and_start(spec) def stop_camera(self, camera): with self.manager._commands: try: sid = self.manager.camera_stream_id(str(camera)) except KeyError: return None group = self.manager.get(sid) if group.spec.get('server_id') != self.server_id: raise ControlError('SERVER_ID_CONFLICT') return self.manager.stop(sid) def authorized(request, token): values = request.headers.getlist('authorization') return (len(values) == 1 and hmac.compare_digest(values[0].encode('utf-8'), ('Bearer '+token).encode('utf-8'))) def build_router(manager, token, server_id, template=None): from app.application.camera_commands import CameraCommands, CommandUnavailable service = ControlService(manager, server_id, template) owner = CameraCommands(service, capacity=16, history=64) def completed(job, future): error = future.exception() LOG.info('secure_command operation=%s camera=%s action=%s outcome=%s error=%s', job.operation_id, job.camera_id, job.action, 'FAILED' if error else 'COMMAND_APPLIED', command_error(error) if error else None) @asynccontextmanager async def lifespan(app): try: yield finally: if not await asyncio.to_thread(owner.close, 2): raise RuntimeError('SECURE_COMMAND_CLEANUP_PENDING') router = APIRouter(lifespan=lifespan) @router.get('/internal/secure-control/runtime-status') async def runtime_status(request: Request): if not authorized(request, token): return JSONResponse({'error': 'CONTROL_UNAUTHORIZED'}, status_code=401) # Parent-only lifecycle snapshots; never query a child collector over IPC. with manager._lock: groups = list(manager._groups.values()) streams = [] counts = dict(running=0, starting=0, stopped=0, error=0, total=len(groups)) for group in groups: s = group.lifecycle.snapshot() observed = s['observed_state'].upper() error = None if observed not in ('RUNNING', 'STARTING', 'STOPPED', 'ERROR'): error, observed = 'TRANSITION_' + observed, 'ERROR' elif s.get('last_error'): error = 'RUNTIME_REPORTED_ERROR' counts[observed.lower()] += 1 streams.append({'cameraId': str(group.spec.get('source_id') or group.stream_id), 'streamId': group.stream_id, 'runId': s['run_id'], 'runGeneration': s['run_generation'], 'desiredState': s['desired_state'].upper(), 'observedState': observed, 'lastError': error}) return {'serverId': server_id, 'status': 'READY', 'modelName': None, 'dependencies': {'runtime': {'ok': True, 'error': None}}, 'streamCounts': counts, 'streams': streams} async def command(request, action): if not authorized(request, token): return JSONResponse({'error':'CONTROL_UNAUTHORIZED'},status_code=401) raw=bytearray() try: async with asyncio.timeout(.25): async for chunk in request.stream(): if len(raw)+len(chunk)>MAX_BODY: return JSONResponse({'error':'BODY_TOO_LARGE'},status_code=413) raw.extend(chunk) except TimeoutError: return JSONResponse({'error':'BODY_TIMEOUT'},status_code=408) try: d = decode(raw) cam = camera_id(d) if action == 'start': validate_start(d) value = Start(json.dumps(d,sort_keys=True,separators=(',',':'),allow_nan=False)) else: if set(d) != {'cameraId'}: raise ControlError('STOP_FIELDS_INVALID') value = None job = owner.submit(cam, value, action) # Register once even when an in-flight duplicate reuses the same operation. if not getattr(job, '_secure_observed', False): job._secure_observed = True LOG.info('secure_command operation=%s camera=%s action=%s outcome=ACCEPTED', job.operation_id, cam, action) job.future.add_done_callback(lambda done: completed(job, done)) # Wait only for admission work, never for GPU/first Frame. Timeout does NOT cancel the job. try: future = asyncio.wrap_future(job.future) # The HTTP budget may end before queued work. Consume later errors without # cancelling work or losing its status in CameraCommands' bounded history. future.add_done_callback(lambda done: None if done.cancelled() else done.exception()) group = await asyncio.wait_for(asyncio.shield(future), .35) if group is None: state = 'STOPPED' else: state = group.observed_state.upper() except asyncio.TimeoutError: state = 'COMMAND_PENDING' except ControlError as exc: return JSONResponse({'error':command_error(exc)},status_code=409) except Exception: return JSONResponse({'error':'RUNTIME_COMMAND_REJECTED'},status_code=409) skipped = ([s['scenarioCode'] for s in d.get('scenarios',[]) if s['enabled'] and s['scenarioCode'] not in ('INTRUSION', 'LOITERING', 'FIRE', 'FALL')]) return JSONResponse({'result':'COMMAND_ACCEPTED','cameraId':cam,'operationId':job.operation_id, 'observedState':state,'skippedScenarios':skipped},status_code=202) except ControlError as exc: return JSONResponse({'error':str(exc)},status_code=422) except CommandUnavailable: return JSONResponse({'error':'CAMERA_COMMAND_BUSY'},status_code=503) @router.post('/internal/secure-control/start') async def start(request:Request): return await command(request,'start') @router.post('/internal/secure-control/stop') async def stop(request:Request): return await command(request,'stop') @router.get('/internal/secure-control/operations/{identifier}') async def status(identifier:str,request:Request): if not authorized(request, token): return JSONResponse({'error':'CONTROL_UNAUTHORIZED'},status_code=401) job=owner.get(identifier) if job is None: return JSONResponse({'error':'NOT_FOUND'},status_code=404) state='PENDING' error=None if job.future.done(): state='FAILED' if job.future.exception() else 'COMMAND_APPLIED' if job.future.exception(): error=command_error(job.future.exception()) return {'operationId':identifier,'cameraId':job.camera_id,'status':state,'error':error} return router