"""Thin, bounded HTTPS/auth relay. Public and Runtime surfaces are separated by Nginx. Nothing is accepted durably here. Event responses are the Manager's responses; 202 queue receipts are never synthesized. No request is retried by this process. """ import asyncio from contextlib import asynccontextmanager import hmac import math import os from contextra_trace.http import Capture, relay_sink from contextra_trace.writer import Recorder from fastapi import FastAPI, Request from fastapi.responses import JSONResponse, Response from control_sidecar.config import Config from control_sidecar.protocol import BoundaryError, loads, bearer, read_body from control_sidecar.transport import Transport from control_sidecar.auth import Auth def validate_event(doc, server_id): if set(doc) != {'serverId', 'events'} or doc['serverId'] != server_id: raise BoundaryError('EVENT_IDENTITY', 422) events = doc['events'] if not isinstance(events, list) or len(events) != 1 or not isinstance(events[0], dict): raise BoundaryError('ONE_EVENT_REQUIRED', 422) e = events[0] required = {'eventId','cameraId','eventType','detectedAtWallMs','startWallTimeMs','rtpTimestamp','score'} allowed = required | {'severityCode','recServerId','recCameraId','clipUrl','thumbnailUrl','detail'} if not required <= set(e) or set(e) - allowed: raise BoundaryError('EVENT_FIELDS_INVALID', 422) if e['eventType'] not in ('INTRUSION', 'LOITERING', 'FIRE', 'FALL', 'FIGHT', 'CROWD'): raise BoundaryError('EVENT_TYPE_UNSUPPORTED', 422) score = e['score'] if type(score) not in (int, float) or not math.isfinite(score) or not 0 <= score <= 1: raise BoundaryError('EVENT_SCORE_INVALID', 422) for k in ('cameraId', 'detectedAtWallMs', 'startWallTimeMs'): if type(e.get(k)) is not int or not 1 <= e[k] <= 2**53-1: raise BoundaryError('EVENT_TIME_OR_ID_INVALID', 422) if type(e.get('rtpTimestamp')) is not int or not 0 <= e['rtpTimestamp'] < 2**32: raise BoundaryError('EVENT_TIME_INVALID', 422) if not isinstance(e.get('eventId'), str) or not e['eventId'].strip() or not 1 <= len(e['eventId']) <= 100: raise BoundaryError('EVENT_ID_INVALID', 422) # New delivery may not advertise insecure media. Stored historical bodies are not rewritten. from urllib.parse import urlsplit for k in ('clipUrl', 'thumbnailUrl'): if e.get(k) is not None: try: if not isinstance(e[k], str) or len(e[k]) > 4096 or any(ord(c) <= 32 for c in e[k]): raise ValueError() p = urlsplit(e[k]) if p.scheme != 'https' or not p.hostname or p.username or p.password or p.query or p.fragment: raise ValueError() _ = p.port except (TypeError, ValueError): raise BoundaryError('MEDIA_HTTPS_REQUIRED', 422) from None def safe_reason(value): import re return value if isinstance(value, str) and re.fullmatch(r'[A-Z][A-Z0-9_]{0,63}', value) else 'RUNTIME_REJECTED' class BoundedRequests: """ASGI-level admission/deadline; cancellation covers the actual downstream task. No queue: at capacity fail promptly. Keep the slot through response completion. A timeout after headers started aborts rather than sending a second response. """ def __init__(self, app, *, seconds, limit): self.app, self.seconds, self.limit = app, seconds, limit self.active = 0 async def __call__(self, scope, receive, send): if scope['type'] != 'http' or scope.get('path') == '/live': return await self.app(scope, receive, send) if self.active >= self.limit: return await JSONResponse({'error': 'BRIDGE_CAPACITY'}, status_code=503)(scope, receive, send) self.active += 1 started = False async def tracked_send(message): nonlocal started if message['type'] == 'http.response.start': started = True await send(message) try: async with asyncio.timeout(self.seconds): await self.app(scope, receive, tracked_send) except TimeoutError: if started: raise await JSONResponse({'error': 'BRIDGE_OUTCOME_UNKNOWN'}, status_code=504)(scope, receive, send) finally: self.active -= 1 def create_app(config=None, transport=None, auth=None): cfg = config or Config.from_env() tx = transport or Transport(cfg.manager_ca) identity = auth or Auth(cfg, tx) from control_sidecar.websocket import ControlSocket sockets = ControlSocket(cfg, tx, identity) trace = None if os.environ.get('CONTEXTRA_TRACE_POLICY_FILE'): trace = Recorder(relay_sink(cfg.runtime+'/internal/trace/ingest',cfg.runtime_token), server=os.environ.get('CONTEXTRA_TRACE_SERVER',cfg.server_id), path=os.environ['CONTEXTRA_TRACE_POLICY_FILE'],start=False) @asynccontextmanager async def lifespan(app): try: if trace: import threading trace.thread = threading.Thread(target=trace.run,name='contextra-trace',daemon=True) trace.thread.start() await identity.start() yield finally: if trace: await asyncio.to_thread(trace.close) await sockets.close() await identity.close() await tx.close() app = FastAPI(lifespan=lifespan, docs_url=None, redoc_url=None, openapi_url=None) app.state.identity = identity app.state.sockets = sockets app.add_api_websocket_route('/ws/v1', sockets.handle) app.add_middleware(BoundedRequests, seconds=cfg.request_seconds, limit=32) if trace: app.add_middleware(Capture,recorder=trace,component='sidecar') @app.get('/internal/trace-status') async def trace_status(request: Request): if not hmac.compare_digest(bearer(request.headers),cfg.runtime_token): return JSONResponse({'error':'TRACE_UNAUTHORIZED'},status_code=401) return trace.status() if trace else {'enabled':False} @app.exception_handler(BoundaryError) async def boundary_error(_, exc): return JSONResponse({'error': exc.code}, status_code=exc.status) @app.get('/live') async def live(): return {'alive': True} @app.get('/ready') async def ready(): identity.current() # This is transport readiness only. GPU/RTSP readiness remains Runtime-owned. return {'ready': True, 'scope': 'AUTH_TRANSPORT_ONLY', 'mode': 'REST_STATELESS'} @app.get('/internal/bridge-status') async def bridge_status(request: Request): if not hmac.compare_digest(bearer(request.headers), cfg.runtime_token): raise BoundaryError('CONTROL_UNAUTHORIZED', 401) return {'auth': identity.failure, 'wss': sockets.snapshot()} @app.get('/internal/media-auth') async def media_auth(request: Request): await identity.authorize(bearer(request.headers)) return Response(status_code=204) async def control(request, action): await identity.authorize(bearer(request.headers)) try: raw = await read_body(request) doc = loads(raw) camera = doc.get('cameraId') if type(camera) is not int or not 0 < camera <= 2**53-1: raise BoundaryError('CAMERA_ID_INVALID', 422) except BoundaryError as error: # Web's business-error envelope is 200; authentication failures remain 401/403. return JSONResponse({'result': 'FAIL', 'code': 'INVALID_REQUEST', 'message': 'Camera request is invalid.', 'reason': error.code}) status, response = await tx.request('POST', cfg.runtime + '/internal/secure-control/' + action, body=raw, token=cfg.runtime_token, seconds=1.0) try: result = loads(response) except BoundaryError: raise BoundaryError('RUNTIME_RESPONSE_INVALID', 502) from None if status == 202 and result.get('result') == 'COMMAND_ACCEPTED' and result.get('cameraId') == camera: state = result.get('observedState') if state == 'ERROR': return JSONResponse({'result': 'FAIL', 'cameraId': camera, 'code': 'INFERENCE_START_FAILED', 'message': 'Runtime reported a lifecycle error.'}) if state not in {'COMMAND_PENDING', 'STARTING', 'RUNNING', 'STOPPING', 'STOPPED'}: raise BoundaryError('RUNTIME_RESPONSE_INVALID', 502) # SECUREBRIDGE-02: bounded queue admission is success, not first Frame. # A genuine upstream timeout still remains unknown; never replay Start here. response = {'result': 'SUCCESS', 'cameraId': camera} if 'requestId' in result: response.update({k: result[k] for k in ('requestId','operationId','observedState','message','change') if k in result}) return JSONResponse(response) if status in (400, 409, 422, 503): failure = {'result': 'FAIL', 'cameraId': camera, 'code': 'INVALID_REQUEST' if status in (400,422) else 'INFERENCE_START_FAILED', 'message': 'Runtime refused the command.', 'reason': safe_reason(result.get('error'))} if 'requestId' in result: failure.update({k: result[k] for k in ('requestId','message','details','existingRunPreserved','nextAction') if k in result}) return JSONResponse(failure) raise BoundaryError('RUNTIME_OUTCOME_UNKNOWN', 502) @app.post('/api/v1/cameras/start') async def start(request: Request): return await control(request, 'start') @app.post('/api/v1/cameras/stop') async def stop(request: Request): return await control(request, 'stop') @app.api_route('/api/v1/maintenance/roi-policy', methods=['GET', 'PATCH']) async def roi_policy(request: Request): await identity.authorize(bearer(request.headers)) raw = await read_body(request) if request.method == 'PATCH' else None status, response = await tx.request(request.method, cfg.runtime+'/internal/secure-control/roi-policy', body=raw, token=cfg.runtime_token, seconds=1.0) return Response(response, status_code=status, media_type='application/json') @app.post('/internal/events') async def events(request: Request): if not hmac.compare_digest(bearer(request.headers), cfg.relay_token): raise BoundaryError('RELAY_UNAUTHORIZED', 401) raw = await read_body(request) validate_event(loads(raw), cfg.server_id) used = identity.current() status, response = await tx.request('POST', cfg.event_endpoint or cfg.manager + '/api/v1/events', body=raw, token=used, seconds=1.5) if status == 401: identity.invalidate(used) # Never resend this POST here. # Preserve bytes/status. Runtime still validates result/accepted/duplicated and owns both attempts. return Response(response, status_code=status, media_type='application/json') return app