"""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 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) @asynccontextmanager async def lifespan(app): try: await identity.start() yield finally: 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) @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. return JSONResponse({'result': 'SUCCESS', 'cameraId': camera}) if status in (400, 409, 422, 503): return JSONResponse({'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'))}) 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.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