indikit.web

The FastAPI web bridge: one shared upstream IndiClient relayed to browsers as typed JSON over a WebSocket, plus a REST snapshot and the bundled panel.

/ws is the whole write surface and /api is a full read of instrument state, so both sit behind the shared token whenever one is configured, and /ws is additionally checked against an Origin allowlist before the handshake is accepted. /health, /, /debug and the static panel stay open. The policy itself is WebSecurity, under Security below.

App

indikit.web.app

FastAPI application factory for the INDIkit web bridge.

:func:create_app builds a FastAPI app that serves, on top of one shared :class:~indikit.client.IndiClient:

• GET / - the built reference panel if present, else the debug inspector page;
• GET /debug - the self-contained debug inspector page;
• GET /health - liveness, the browser contract's version, upstream connection state, and counters for the slow-sink drops and the upstream parser;
• GET /api/devices and /api/devices/{device}[/{name}] - a read-only JSON snapshot of the property cache;
• WS /ws - the live bridge: a snapshot on connect, then streamed updates, with browser-sent frames forwarded upstream.

The client is injectable so tests drive the app over an in-memory transport; by default a real TCP client to indiserver is created.

The auth boundary runs around /ws and /api. /ws is the write surface, and /api/devices/* is a full read of instrument state - site coordinates, hardware inventory, mount and focuser positions - so both sit behind the token whenever one is configured. /health stays open because the image's HEALTHCHECK calls it unauthenticated and it exposes one boolean and a handful of counters - no addresses, no device names, and no release version, which would hand an unauthenticated caller the exact build to look up advisories against while telling a legitimate one nothing the protocol integer does not already say; /, /debug and the static panel are open-source HTML with nothing instrument-specific in them. There is deliberately no ambient credential (no cookie, no session), so cross-origin JavaScript cannot authenticate to /api at all and there are no state-changing HTTP routes to protect; see :mod:indikit.web.security for what guards /ws.

create_app

create_app(*, client: IndiClient | None = None, indi_host: str = 'localhost', indi_port: int = 7624, token: str | None = None, allowed_origins: Sequence[str] = (), connect_timeout: float = 10.0, reconnect_delay: float = 2.0, message_history: int = _MESSAGE_HISTORY, max_backlog: int = _MAX_BACKLOG) -> FastAPI

Build the web-bridge FastAPI application.

Nothing here reads the environment. indikit serve fills the tuning arguments from :class:~indikit.settings.Settings, so the app stays injectable and importing it costs no ambient configuration.

Parameters:

Name	Type	Description	Default
client	IndiClient	An existing client to relay (used by tests); if omitted, a real :class:IndiClient to indi_host/indi_port is created.	None
indi_host	str	Upstream indiserver host (when client is not given).	'localhost'
indi_port	int	Upstream indiserver port (when client is not given).	7624
connect_timeout	float	Seconds the created client waits per connection attempt (when client is not given).	10.0
reconnect_delay	float	Seconds the created client waits between attempts (when client is not given).	2.0
message_history	int	How many recent INDI message frames the bridge replays to a newly attached browser.	_MESSAGE_HISTORY
max_backlog	int	How many live frames a browser may fall behind by before it is dropped.	_MAX_BACKLOG
token	str	A shared secret required on /ws and /api. None (the default) leaves both open, which is what a loopback-bound development server wants.	None
allowed_origins	Sequence of str	Browser origins accepted on /ws in addition to the server's own, e.g. http://localhost:5173 for a Vite dev server or a separate front end. "*" accepts any.	()

Returns:

Name	Type	Description
app	FastAPI	The configured application; its lifespan starts and stops the bridge.

Source code in src/indikit/web/app.py

def create_app(
    *,
    client: IndiClient | None = None,
    indi_host: str = "localhost",
    indi_port: int = 7624,
    token: str | None = None,
    allowed_origins: Sequence[str] = (),
    connect_timeout: float = 10.0,
    reconnect_delay: float = 2.0,
    message_history: int = _MESSAGE_HISTORY,
    max_backlog: int = _MAX_BACKLOG,
) -> FastAPI:
    """Build the web-bridge FastAPI application.

    Nothing here reads the environment. ``indikit serve`` fills the tuning
    arguments from :class:`~indikit.settings.Settings`, so the app stays
    injectable and importing it costs no ambient configuration.

    Parameters
    ----------
    client : IndiClient, optional
        An existing client to relay (used by tests); if omitted, a real
        :class:`IndiClient` to ``indi_host``/``indi_port`` is created.
    indi_host : str, optional
        Upstream ``indiserver`` host (when ``client`` is not given).
    indi_port : int, optional
        Upstream ``indiserver`` port (when ``client`` is not given).
    connect_timeout : float, optional
        Seconds the created client waits per connection attempt (when ``client``
        is not given).
    reconnect_delay : float, optional
        Seconds the created client waits between attempts (when ``client`` is not
        given).
    message_history : int, optional
        How many recent INDI ``message`` frames the bridge replays to a newly
        attached browser.
    max_backlog : int, optional
        How many live frames a browser may fall behind by before it is dropped.
    token : str, optional
        A shared secret required on ``/ws`` and ``/api``. `None` (the default)
        leaves both open, which is what a loopback-bound development server
        wants.
    allowed_origins : Sequence of str, optional
        Browser origins accepted on ``/ws`` in addition to the server's own,
        e.g. ``http://localhost:5173`` for a Vite dev server or a separate
        front end. ``"*"`` accepts any.

    Returns
    -------
    app : FastAPI
        The configured application; its lifespan starts and stops the bridge.
    """
    indi_client = client or IndiClient(
        indi_host,
        indi_port,
        connect_timeout=connect_timeout,
        reconnect_delay=reconnect_delay,
    )
    bridge = Bridge(indi_client, message_history=message_history, max_backlog=max_backlog)
    security = WebSecurity.build(token, allowed_origins)

    @asynccontextmanager
    async def lifespan(_: FastAPI) -> AsyncIterator[None]:
        """Start the bridge on startup and close it on shutdown."""
        await bridge.start()
        try:
            yield
        finally:
            await bridge.aclose()

    app = FastAPI(title="INDIkit web bridge", lifespan=lifespan)
    app.state.bridge = bridge
    app.state.client = indi_client
    app.state.security = security

    def require_token(request: Request) -> None:
        """Reject a request that does not carry the configured token.

        ``Authorization: Bearer`` only; see :func:`_ws_token` for why ``?token=``
        stops at the WebSocket handshake.

        Parameters
        ----------
        request : Request
            The incoming request.

        Raises
        ------
        HTTPException
            Raised with 403 when a token is configured and not supplied.
        """
        if not security.token_ok(_bearer_token(request)):
            raise HTTPException(status_code=403, detail="a valid token is required")

    api = [Depends(require_token)]

    @app.get("/health")
    async def health() -> dict[str, Any]:
        """Report liveness, the upstream link, the parser, and dropped browsers.

        Open on purpose: the container's ``HEALTHCHECK`` calls it with no
        credentials. ``dropped_slow_sinks`` is here because a browser cannot tell
        an overflow drop from a network fault, so the count has to be readable
        from the outside without scraping logs.

        **The body only ever grows.** ``status``, ``connected`` and
        ``dropped_slow_sinks`` keep their names at the top level, because a
        monitoring check somewhere is already reading them; everything since was
        added beside them rather than by nesting or renaming those three.

        ``protocol`` is the browser contract's version
        (:data:`~indikit.web.control_frames.BRIDGE_PROTOCOL_VERSION`), the
        same integer the ``hello`` frame carries. It is here so a deployment
        check can answer "will my pinned client understand this bridge" without
        opening a WebSocket. The release version is deliberately **not** here;
        see this module's docstring.

        ``coalesced_blobs`` counts images that were replaced in a browser's queue
        before it read them. The bridge delivers the latest exposure rather than
        every exposure, which is the only bounded thing it can do
        (:meth:`~indikit.web.bridge.Subscription.enqueue` argues it), and this
        is where that shows up instead of being silent. It is additive and **not**
        a ``protocol`` bump: nothing on the browser contract changed.

        While disconnected, ``upstream.uptime_seconds`` and
        ``last_message_age_seconds`` are ``null`` and ``reconnects`` keeps its
        count, while the ``parser`` block reports the **last** connection's final
        counters - stated rather than left to be discovered, because a frozen
        ``bytes_since_last_message`` on a dead link is exactly the field an
        operator would misread. The two ``_total`` fields are the durable ones.
        """
        stats = indi_client.stats
        return {
            "status": "ok",
            "protocol": BRIDGE_PROTOCOL_VERSION,
            "connected": stats.connected,
            "dropped_slow_sinks": bridge.dropped_slow_sinks,
            "coalesced_blobs": bridge.coalesced_blobs,
            "sinks_attached": bridge.sink_count(),
            "upstream": {
                "uptime_seconds": _seconds(stats.uptime_seconds),
                "reconnects": stats.reconnects,
                "last_message_age_seconds": _seconds(stats.last_message_age_seconds),
            },
            "parser": {
                "dropped": stats.dropped,
                "resets": stats.resets,
                "bytes_since_last_message": stats.bytes_since_last_message,
                "dropped_total": stats.dropped_total,
                "resets_total": stats.resets_total,
            },
        }

    @app.get("/api/devices", dependencies=api)
    async def list_devices() -> list[str]:
        """Return the names of all known devices."""
        return indi_client.store.devices()

    @app.get("/api/devices/{device}", dependencies=api)
    async def device_properties(device: str) -> dict[str, Vector]:
        """Return one device's properties as JSON, keyed by property name.

        A known device that currently publishes nothing returns ``{}``, not a
        404. It is a device that has retracted its properties - a driver that
        defines them on connect, seen while disconnected - and it is still
        listed by ``/api/devices``, so answering "unknown" here would have the
        two endpoints contradict each other.

        The annotation is the point: FastAPI serialises through it, so the
        payload is pinned by the same ``Vector`` schema the WebSocket carries
        and OpenAPI documents it as one instead of as a bare object. It is not
        free - a response model re-validates on the way out, so a large cache
        costs N validations per request - and this is the right endpoint to pay
        it on, being a snapshot rather than the live stream.
        """
        if device not in indi_client.store:
            raise HTTPException(status_code=404, detail=f"unknown device {device!r}")
        return dict(indi_client.store.device(device))

    @app.get("/api/devices/{device}/{name}", dependencies=api)
    async def one_property(device: str, name: str) -> Vector:
        """Return a single property vector as JSON."""
        vec = indi_client.store.get(device, name)
        if vec is None:
            raise HTTPException(status_code=404, detail=f"unknown property {device}.{name}")
        return vec

    async def _receive_loop(websocket: WebSocket, sub: Subscription) -> None:
        """Forward this browser's frames upstream until it disconnects.

        Parameters
        ----------
        websocket : WebSocket
            The browser's socket.
        sub : Subscription
            The browser's bridge subscription, so a rejected frame is reported
            back to it alone.
        """
        with contextlib.suppress(WebSocketDisconnect):
            while True:
                await bridge.handle_incoming(await websocket.receive_text(), sub)

    @app.websocket("/ws")
    async def ws(websocket: WebSocket) -> None:
        """Stream live updates to a browser and forward its frames upstream.

        The route sends nothing itself: seeding and registration are one
        synchronous operation inside :meth:`Bridge.attach`, so no event can be
        lost in the window between them.
        """
        headers = websocket.headers
        if not security.origin_allowed(headers.get("origin"), headers.get("host")):
            # Closing before accept() answers the handshake with an HTTP error
            # rather than opening a socket and then dropping it.
            await websocket.close(code=_WS_POLICY_VIOLATION)
            return
        if not security.token_ok(_ws_token(websocket)):
            await websocket.close(code=_WS_POLICY_VIOLATION)
            return
        await websocket.accept()
        sub = bridge.attach(websocket.send_text)
        receiving = asyncio.create_task(_receive_loop(websocket, sub))
        # Racing the drop matters: a browser the bridge dropped for backlog would
        # otherwise leave this route parked in receive_text() with no pump behind
        # it, holding a socket that will never be served again.
        dropped = asyncio.create_task(sub.closed.wait())
        try:
            await asyncio.wait({receiving, dropped}, return_when=asyncio.FIRST_COMPLETED)
        finally:
            receiving.cancel()
            dropped.cancel()
            # Awaited one at a time rather than through gather(): a cancelled
            # child's CancelledError comes up out of gather() even under
            # return_exceptions, which cancelled this route mid-cleanup. Reaping
            # them is not optional either - without it a cancellation that has
            # not been delivered yet surfaces later as "Task was destroyed but it
            # is pending!", and sub.aclose() only happens to give the loop enough
            # turns.
            for task in (receiving, dropped):
                with contextlib.suppress(asyncio.CancelledError):
                    await task
            # Only a receive loop that already finished can have failed, and
            # that is the one whose exception nothing else would ever report.
            if receiving.done() and not receiving.cancelled():
                failure = receiving.exception()
                if failure is not None:
                    logger.error("websocket receive loop failed", exc_info=failure)
            await sub.aclose()
            # The backlog flag, not `sub.closed`: every way out of the pump sets
            # the event, so keying the code off it told an ordinary disconnect
            # racing a stale write failure to "try again later".
            code = _WS_TRY_AGAIN_LATER if sub.dropped_for_backlog else _WS_NORMAL_CLOSURE
            with contextlib.suppress(RuntimeError):
                await websocket.close(code=code)

    @app.get("/debug")
    async def debug_page() -> FileResponse:
        """Serve the self-contained debug inspector page."""
        return FileResponse(_STATIC / "debug.html")

    # Serve the built reference panel at the root when it is present (produced by
    # ``pnpm --filter @indikit/panel build``); otherwise fall back to the debug
    # page. The static mount is added last so the API/WS/debug routes above win.
    if (_PANEL / "index.html").is_file():
        app.mount("/", StaticFiles(directory=_PANEL, html=True), name="panel")
    else:

        @app.get("/")
        async def index() -> FileResponse:
            """Serve the debug page when the reference panel is not built."""
            return FileResponse(_STATIC / "debug.html")

    return app

Bridge

indikit.web.bridge

Bridge: fan one shared IndiClient out to many browser WebSockets.

The bridge owns a single upstream connection to indiserver (via the M3 :class:~indikit.client.IndiClient) and relays its activity to every connected browser as JSON. Property changes, log messages, and connection-state transitions are broadcast to every attached browser; a browser's inbound frames are parsed back into typed models and forwarded upstream. A newly-attached browser is first seeded with the current cache so it starts with full state.

Server -> browser frames are either an INDI message ({"tag": ...}, mirroring the protocol models) or a small bridge control frame ({"event": ...}, modelled in :mod:indikit.web.control_frames) for things the INDI protocol has no message for: the version of the browser contract, upstream connection state, and the rejection of a frame the browser sent. The first frame on every socket is the hello, ahead of the seeded properties, so a browser knows what it is talking to before it has to interpret anything. Browser -> server frames are always INDI messages, and only the three a client is allowed to send.

A browser is a subscriber, never something the upstream reader awaits. Every broadcast originates in :class:IndiClient's single connection task - property events and messages through the reader, the connection frame through the connection loop - so awaiting a socket from :meth:Bridge._broadcast would let one browser under TCP back-pressure stall the upstream stream for everyone, eventually stalling the parser and tearing down a healthy connection. Instead each browser gets a bounded queue and its own pump task, and :meth:Bridge._broadcast is a plain synchronous def that appends and returns.

That same synchrony is what makes seeding atomic: :meth:Bridge.attach reads the cache, builds the seed and registers the subscriber with no await anywhere between, so no event can land in the window that used to exist between the snapshot and the registration. Keep attach and _broadcast synchronous - that is the whole argument, and a refactor that quietly makes either one a coroutine reopens the hole.

Subscription dataclass

Subscription(bridge: Bridge, sink: Sink, seed_vectors: tuple[Vector, ...], seed_frames: tuple[str, ...], preamble: tuple[str, ...], queue: deque[_Slot] = deque(), coalescible: dict[tuple[str, str], _Slot] = dict(), ready: Event = asyncio.Event(), closed: Event = asyncio.Event(), dropped_for_backlog: bool = False, task: Task[None] | None = None)

One attached browser: its seed, its queued frames, and its pump task.

Returned by :meth:Bridge.attach and closed by :meth:aclose. Not constructed directly.

Attributes:

Name	Type	Description
bridge	Bridge	The bridge this subscription is attached to.
sink	Sink	The awaitable that sends one text frame to this browser.
seed_vectors	tuple of Vector	The cached properties to send first, held as references rather than as serialized JSON so N attaching browsers do not buffer N copies of the cache before a byte drains. They are serialized by the pump, at drain rate.
seed_frames	tuple of str	The retained message frames plus the connection frame, sent after the seeded properties.
preamble	tuple of str	Frames sent before the seeded properties: the hello, which has to be the first thing on the socket because it says which contract everything after it is written in.
queue	deque of _Slot	Live frames queued since the attach, drained after the seed.
coalescible	dict	(device, name) -> the queued set slot a later set may replace.
ready	Event	Set when queue is non-empty; the pump waits on it.
closed	Event	Set once the pump has exited, however it exited.
dropped_for_backlog	bool	Set by :meth:Bridge._drop alone, and it is the reason rather than the fact: closed is set by every path out of the pump, so a browser that simply went away while a stale write was failing looks identical through it. The route reads this to choose its close code, and telling a browser "try again later" for an ordinary disconnect is a lie about why its socket ended.
task	Task or None	The pump; assigned by :meth:Bridge.attach immediately after construction.

send_control

send_control(frame: str) -> None

Queue a bridge control frame for this browser alone.

Parameters:

Name	Type	Description	Default
frame	str	The JSON control frame to send.	required

Source code in src/indikit/web/bridge.py

def send_control(self, frame: str) -> None:
    """Queue a bridge control frame for this browser alone.

    Parameters
    ----------
    frame : str
        The JSON control frame to send.
    """
    self.bridge._deliver(self, frame, None)

enqueue

enqueue(frame: str, key: tuple[str, str] | None) -> bool

Append a frame, or fold it into the queued frame for the same key.

Parameters:

Name	Type	Description	Default
frame	str	The JSON text to send.	required
key	tuple of str or None	The (device, name) this frame may coalesce onto. None always appends.	required

Returns:

Name	Type	Description
replaced	bool	Whether this frame overwrote a queued one rather than joining the queue. The caller counts that for BLOBs, where a dropped frame is a whole exposure the browser will never see.

Source code in src/indikit/web/bridge.py

def enqueue(self, frame: str, key: tuple[str, str] | None) -> bool:
    """Append a frame, or fold it into the queued frame for the same key.

    Parameters
    ----------
    frame : str
        The JSON text to send.
    key : tuple of str or None
        The ``(device, name)`` this frame may coalesce onto. `None` always
        appends.

    Returns
    -------
    replaced : bool
        Whether this frame overwrote a queued one rather than joining the
        queue. The caller counts that for BLOBs, where a dropped frame is a
        whole exposure the browser will never see.
    """
    if key is not None:
        queued = self.coalescible.get(key)
        if queued is not None:
            # Replacing in place keeps the frame's position in the queue, so
            # ordering against everything else is untouched.
            #
            # BLOBs coalesce along with everything else, and that is a
            # deliberate choice rather than a consequence of `set` being
            # last-writer-wins state. Three reasons, none of them optional:
            #
            # 1. The bridge could not be lossless for BLOBs even if it tried.
            #    `PropertyStore` overwrites the payload in the cached vector
            #    in place, so once the next exposure has been folded in there
            #    is nothing behind the queue left to replay.
            # 2. `_MAX_BACKLOG` counts frames, not bytes. Queueing every
            #    image removes the memory bound outright: thirty 8 MiB frames
            #    already measure at 372 MiB, and a browser at the cap projects
            #    to gigabytes - the process dies before `dropped_slow_sinks`
            #    can even record a drop.
            # 3. INDI 1.7 licenses it explicitly: a server may drop BLOBs
            #    arriving faster than a slow recipient takes them, and must
            #    not block while writing a large BLOB to one.
            #
            # A consumer that needs every exposure wants a Python
            # `IndiClient` on TCP, which does not coalesce; see
            # `examples/blob_receiver.py`.
            queued.frame = frame
            return True
    slot = _Slot(frame, key)
    self.queue.append(slot)
    if key is not None:
        self.coalescible[key] = slot
    self.ready.set()
    return False

invalidate

invalidate(device: str, name: str | None) -> None

Forget the coalescible slot(s) for a property, or for a whole device.

Called for every def and del: a later set folded into a slot sitting ahead of a queued retraction or redefinition would overtake it, and the browser would apply them out of order. This is the correctness condition the whole coalescing rests on.

Parameters:

Name	Type	Description	Default
device	str	The device whose queued set frames may no longer be replaced.	required
name	str or None	The property, or None for a whole-device delProperty, which takes every property the device had.	required

Source code in src/indikit/web/bridge.py

def invalidate(self, device: str, name: str | None) -> None:
    """Forget the coalescible slot(s) for a property, or for a whole device.

    Called for every ``def`` and ``del``: a later ``set`` folded into a slot
    sitting *ahead* of a queued retraction or redefinition would overtake it,
    and the browser would apply them out of order. This is the correctness
    condition the whole coalescing rests on.

    Parameters
    ----------
    device : str
        The device whose queued ``set`` frames may no longer be replaced.
    name : str or None
        The property, or `None` for a whole-device ``delProperty``, which
        takes every property the device had.
    """
    if name is None:
        for key in [key for key in self.coalescible if key[0] == device]:
            del self.coalescible[key]
        return
    self.coalescible.pop((device, name), None)

pop

pop() -> str

Remove and return the next queued frame.

Returns:

Name	Type	Description
frame	str	The JSON text to send.

Source code in src/indikit/web/bridge.py

def pop(self) -> str:
    """Remove and return the next queued frame.

    Returns
    -------
    frame : str
        The JSON text to send.
    """
    slot = self.queue.popleft()
    if slot.key is not None and self.coalescible.get(slot.key) is slot:
        del self.coalescible[slot.key]
    return slot.frame

aclose async

aclose() -> None

Stop this subscription's pump and deregister it. Idempotent.

Source code in src/indikit/web/bridge.py

async def aclose(self) -> None:
    """Stop this subscription's pump and deregister it. Idempotent."""
    if self.task is None:
        return
    self.task.cancel()
    with contextlib.suppress(asyncio.CancelledError):
        await self.task
    # Not only the pump's `finally`: a task cancelled before its first step
    # never runs its body at all, so the bookkeeping has to happen here too.
    self.bridge._detach(self)

Bridge

Bridge(client: IndiClient, server: str = __version__, *, message_history: int = _MESSAGE_HISTORY, max_backlog: int = _MAX_BACKLOG)

Relay between one IndiClient and many browser WebSocket sinks.

Parameters:

Name	Type	Description	Default
client	IndiClient	The shared upstream client the bridge relays.	required
server	str	The server version stamped into every socket's hello frame; defaults to :data:indikit.__version__. A parameter rather than a lookup at the send site so a test can pin it, and defaulted rather than required because every caller in and out of this repository wants the real one.	__version__
message_history	int	How many recent INDI message frames to replay to a newly attached browser; defaults to :data:_MESSAGE_HISTORY.	_MESSAGE_HISTORY
max_backlog	int	How many live frames a browser may fall behind by before it is dropped; defaults to :data:_MAX_BACKLOG.	_MAX_BACKLOG

Source code in src/indikit/web/bridge.py

def __init__(
    self,
    client: IndiClient,
    server: str = __version__,
    *,
    message_history: int = _MESSAGE_HISTORY,
    max_backlog: int = _MAX_BACKLOG,
) -> None:
    self._client = client
    self._hello = dump_frame(HelloFrame(server=server))
    self._subs: set[Subscription] = set()
    self._dropped_slow_sinks = 0
    self._coalesced_blobs = 0
    self._max_backlog = max_backlog
    # INDI messages are transient (not part of the property cache), so keep a
    # bounded history to prime a newly-attached browser's log.
    self._messages: deque[str] = deque(maxlen=message_history)

client property

client: IndiClient

The upstream client this bridge relays.

dropped_slow_sinks property

dropped_slow_sinks: int

How many browsers have been dropped for falling too far behind.

Surfaced on /health because a browser cannot tell an overflow drop from a network fault - both are a closed socket it reconnects from - so the diagnosis has to be readable server-side without scraping logs.

coalesced_blobs property

coalesced_blobs: int

How many queued BLOB frames were replaced before a browser read them.

One per browser per skipped image: the bridge delivers the latest exposure, not every exposure (see :meth:Subscription.enqueue for why), and this is the only place that shows it happening. A rising count on an otherwise healthy bridge says a browser is being served images slower than the camera produces them, which is a fact about the socket and not a fault.

Only BLOB coalescing is counted. A temperature readout coalesces constantly and by design, so a count over every frame kind would run away from the first minute and tell an operator nothing.

hello_frame property

hello_frame: str

The hello frame this bridge leads every socket with.

Built once at construction: it is the same text for every browser, and the version it carries cannot change while the process runs.

start async

start() -> None

Subscribe to the client and open its upstream connection.

Returns once the connection attempt is under way rather than once it succeeds: the bridge has to come up whether or not indiserver is reachable, so a browser gets the panel and a disconnected indicator instead of a server that never finishes starting. The client keeps retrying in the background and announces the connection when it lands.

Source code in src/indikit/web/bridge.py

async def start(self) -> None:
    """Subscribe to the client and open its upstream connection.

    Returns once the connection attempt is under way rather than once it
    succeeds: the bridge has to come up whether or not ``indiserver`` is
    reachable, so a browser gets the panel and a disconnected indicator
    instead of a server that never finishes starting. The client keeps
    retrying in the background and announces the connection when it lands.
    """
    self._client.subscribe(self._on_event)
    self._client.on_message(self._on_message)
    self._client.on_connection(self._on_connection)
    await self._client.start(wait=False)

aclose async

aclose() -> None

Stop every pump and close the upstream connection.

A pump cancelled mid-send may leave a half-written socket; this is shutdown, and the route that owns the socket closes it in its own finally.

Source code in src/indikit/web/bridge.py

async def aclose(self) -> None:
    """Stop every pump and close the upstream connection.

    A pump cancelled mid-send may leave a half-written socket; this is
    shutdown, and the route that owns the socket closes it in its own
    ``finally``.
    """
    subs = list(self._subs)
    for sub in subs:
        if sub.task is not None:
            sub.task.cancel()
    await asyncio.gather(
        *(sub.task for sub in subs if sub.task is not None), return_exceptions=True
    )
    await self._client.aclose()

attach

attach(sink: Sink) -> Subscription

Register a browser and seed it with the current cache.

Synchronous on purpose, and it must stay that way. Reading the cache, building the seed and joining the subscriber set happen with no await between them, so the client's task cannot run in the middle and no event can be lost between the snapshot and the registration. create_task only schedules, so the pump has not started when this returns.

The seed holds vector references. The property store merges a set into the cached vector in place, so a definition serialized at pump time may show a newer value than the property had at attach time. That converges rather than corrupts: the queued set frames were serialized eagerly, in order, and the last one always equals the current object, so the browser ends on the right value and only skips intermediate ones - which is what INDI set means anyway. A property deleted in the window arrives as its definition followed by the queued delProperty, in that order.

The hello frame leads, ahead of the seeded properties: it names the contract version everything after it is written in, so it cannot follow the frames a browser would need it to interpret.

Parameters:

Name	Type	Description	Default
sink	Sink	An awaitable that sends one text frame to the browser.	required

Returns:

Name	Type	Description
subscription	Subscription	The browser's handle; close it with :meth:Subscription.aclose when the socket goes away.

Source code in src/indikit/web/bridge.py

def attach(self, sink: Sink) -> Subscription:
    """Register a browser and seed it with the current cache.

    **Synchronous on purpose, and it must stay that way.** Reading the cache,
    building the seed and joining the subscriber set happen with no ``await``
    between them, so the client's task cannot run in the middle and no event
    can be lost between the snapshot and the registration. ``create_task``
    only schedules, so the pump has not started when this returns.

    The seed holds vector *references*. The property store merges a ``set``
    into the cached vector in place, so a definition serialized at pump time
    may show a newer value than the property had at attach time. That
    converges rather than corrupts: the queued ``set`` frames were serialized
    eagerly, in order, and the last one always equals the current object, so
    the browser ends on the right value and only skips intermediate ones -
    which is what INDI ``set`` means anyway. A property deleted in the window
    arrives as its definition followed by the queued ``delProperty``, in that
    order.

    The ``hello`` frame leads, ahead of the seeded properties: it names the
    contract version everything after it is written in, so it cannot follow
    the frames a browser would need it to interpret.

    Parameters
    ----------
    sink : Sink
        An awaitable that sends one text frame to the browser.

    Returns
    -------
    subscription : Subscription
        The browser's handle; close it with
        :meth:`Subscription.aclose` when the socket goes away.
    """
    store = self._client.store
    sub = Subscription(
        bridge=self,
        sink=sink,
        seed_vectors=tuple(
            vector for device in store.devices() for vector in store.device(device).values()
        ),
        seed_frames=(*self._messages, self.connection_frame(self._client.connected)),
        preamble=(self._hello,),
    )
    self._subs.add(sub)
    sub.task = asyncio.create_task(self._pump(sub), name=f"bridge-sink-{id(sub):x}")
    return sub

sink_count

sink_count() -> int

Return how many browsers are currently attached.

Returns:

Name	Type	Description
count	int	The number of live subscriptions.

Source code in src/indikit/web/bridge.py

def sink_count(self) -> int:
    """Return how many browsers are currently attached.

    Returns
    -------
    count : int
        The number of live subscriptions.
    """
    return len(self._subs)

connection_frame staticmethod

connection_frame(connected: bool) -> str

Build the control frame announcing upstream connection state.

Parameters:

Name	Type	Description	Default
connected	bool	Whether the bridge is connected to indiserver.	required

Returns:

Name	Type	Description
frame	str	A JSON control frame ({"event": "connection", ...}).

Source code in src/indikit/web/bridge.py

@staticmethod
def connection_frame(connected: bool) -> str:
    """Build the control frame announcing upstream connection state.

    Parameters
    ----------
    connected : bool
        Whether the bridge is connected to ``indiserver``.

    Returns
    -------
    frame : str
        A JSON control frame (``{"event": "connection", ...}``).
    """
    return dump_frame(ConnectionFrame(connected=connected))

handle_incoming async

handle_incoming(text: str, sub: Subscription | None = None) -> None

Parse a browser frame and forward it upstream.

A frame that cannot be parsed, that a client is not allowed to send, or that the upstream refuses is reported back to the browser that sent it as an {"event": "error"} control frame, and the socket stays open. Silence would be a regression: sends used to be queued for a later connection, so a browser that hears nothing has no reason not to assume its write landed.

Parameters:

Name	Type	Description	Default
text	str	A JSON INDI message from the browser.	required
sub	Subscription	The sender, so a rejection reaches it alone. None (a caller with no socket, e.g. a test) logs instead.	None

Source code in src/indikit/web/bridge.py

async def handle_incoming(self, text: str, sub: Subscription | None = None) -> None:
    """Parse a browser frame and forward it upstream.

    A frame that cannot be parsed, that a client is not allowed to send, or
    that the upstream refuses is reported back to the browser that sent it as
    an ``{"event": "error"}`` control frame, and the socket stays open.
    Silence would be a regression: sends used to be queued for a later
    connection, so a browser that hears nothing has no reason not to assume
    its write landed.

    Parameters
    ----------
    text : str
        A JSON INDI message from the browser.
    sub : Subscription, optional
        The sender, so a rejection reaches it alone. `None` (a caller with
        no socket, e.g. a test) logs instead.
    """
    try:
        msg = from_json(text)
    except (ValueError, TypeError):
        logger.warning("dropping malformed inbound frame: %r", text[:200])
        self._reject(sub, "malformed", "frame is not a valid INDI message", None)
        return
    if not isinstance(msg, _CLIENT_TO_SERVER):
        logger.warning("dropping browser frame a client may not send: %s", type(msg).__name__)
        self._reject(sub, "not_permitted", "a client may not send this message", msg.tag)
        return
    try:
        if isinstance(msg, EnableBLOB):
            # Through enable_blob, not send: the client records the policy
            # there and replays it on every reconnect, so a browser's BLOB
            # subscription survives an upstream restart.
            await self._client.enable_blob(msg.device, msg.name, msg.policy)
        else:
            await self._client.send(msg)
    except NotConnectedError:
        # Named, not ConnectionError/OSError: both new types keep a builtin
        # base, and catching the base would swallow unrelated failures.
        detail = (
            "not connected to indiserver; the policy is stored and will apply on reconnect"
            if isinstance(msg, EnableBLOB)
            else "not connected to indiserver; the write was not sent"
        )
        self._reject(sub, "not_connected", detail, msg.tag)
    except SendQueueFull:
        logger.warning("upstream outbox full; dropping %s from a browser", type(msg).__name__)
        self._reject(
            sub, "upstream_busy", "the upstream queue is full; the write was not sent", msg.tag
        )

Control frames

The non-INDI half of the browser contract: hello, connection and error, plus BRIDGE_PROTOCOL_VERSION, the version of that contract. See Protocol concepts for what a client does with a version mismatch.

indikit.web.control_frames

The bridge's control frames: the non-INDI half of the browser contract.

Everything the bridge sends a browser is either an INDI message ({"tag": ...}, from :mod:indikit.protocol.models) or one of the frames here ({"event": ...}), which exist for the things INDI 1.7 has no message for: the version of the browser contract itself, the upstream connection state, and the rejection of a frame the browser sent.

They are models rather than hand-built dictionaries for the same reason the INDI messages are: web/packages/client/src/types.ts is a hand-authored mirror of the Python models, and a frame assembled with json.dumps at three call sites has no schema for that mirror to be checked against. :data:BridgeFrame is discriminated on event exactly as IndiMessage is on tag, so the union is closed and tests/test_wire_contract.py can snapshot its schema.

Versioning. :data:BRIDGE_PROTOCOL_VERSION versions this contract - the JSON a browser sees - and has nothing to do with INDI's own version attribute on getProperties, which is frozen at 1.7. It is announced once per socket in the :class:HelloFrame, ahead of every other frame, so a browser learns what it is talking to before it has to interpret anything. A mismatch is never fatal in either direction: INDI has always been additive-tolerant, the client drops an event it does not know, and turning a cosmetic version skew into a dark panel mid-session helps nobody.

HelloFrame

Bases: _Frame

The first frame on every /ws socket: what the browser is talking to.

Attributes:

Name	Type	Description
event	str	Always "hello".
protocol	int	The browser contract's version; see :data:BRIDGE_PROTOCOL_VERSION.
server	str	The INDIkit version serving this socket, for a UI to display and for a bug report to quote. It has no model default on purpose: a default lands in model_json_schema(), so pinning indikit.__version__ here would break the golden wire schema on every release. The bridge supplies it at construction instead.

ConnectionFrame

Bases: _Frame

The upstream indiserver link went up or down.

Attributes:

Name	Type	Description
event	str	Always "connection".
connected	bool	Whether the bridge currently has a live upstream connection.

ErrorFrame

Bases: _Frame

A frame this browser sent did not go upstream.

Sent to that browser alone, never for something the bridge accepted, and the socket stays open. Silence would be worse: a refused write is not retried anywhere, so a browser that hears nothing has no reason not to believe it landed.

Attributes:

Name	Type	Description
event	str	Always "error".
code	str	A stable machine-readable reason, e.g. "not_connected".
message	str	Human-readable detail, suitable for a UI log.
tag	str or None	The rejected message's INDI tag, or None if it never parsed.

dump_frame

dump_frame(frame: BridgeFrame) -> str

Serialise one control frame to the JSON text a browser receives.

Parameters:

Name	Type	Description	Default
frame	BridgeFrame	The frame to serialise.	required

Returns:

Name	Type	Description
text	str	The frame as compact JSON, matching what :func:indikit.protocol.to_json produces for an INDI message.

Source code in src/indikit/web/control_frames.py

def dump_frame(frame: BridgeFrame) -> str:
    """Serialise one control frame to the JSON text a browser receives.

    Parameters
    ----------
    frame : BridgeFrame
        The frame to serialise.

    Returns
    -------
    text : str
        The frame as compact JSON, matching what
        :func:`indikit.protocol.to_json` produces for an INDI message.
    """
    return frame.model_dump_json()

Security

indikit.web.security

The access controls in front of the web bridge: an origin allowlist and a token.

WS /ws is the bridge's entire write surface: a frame arriving there becomes an INDI new* message on the upstream connection, which is how a browser slews a mount or opens a shutter. Browsers do not apply the same-origin policy to WebSockets and CORS does not cover them either, so without a check here any page an operator happens to visit can open ws://localhost:8000/ws and drive the instrument - cross-site WebSocket hijacking. The Origin check is therefore the control on that surface, not defence in depth on top of one.

Two deliberate decisions live in :meth:WebSecurity.origin_allowed:

• A missing Origin is allowed. A browser always sends one on a WebSocket handshake, so refusing a request without one stops no browser; what it does stop is every non-browser peer - a Node consumer of @indikit/client, curl, the interop suite, Starlette's own TestClient - none of which sends the header. An attacker outside a browser sets the header to anything it likes, so the requirement would cost real users and buy nothing.
• X-Forwarded-* is not consulted. A header any client can forge is not an authorization input. Behind a reverse proxy that rewrites Host, name the browser-facing origin explicitly with --allow-origin.

Everything here is a pure function over header strings, so it is testable without a FastAPI request and has no import cycle with :mod:indikit.web.app.

WebSecurity dataclass

WebSecurity(token: str | None = None, allowed_origins: frozenset[str] = frozenset())

The bridge's access policy: which origins may connect, and with what token.

Parameters:

Name	Type	Description	Default
token	str or None	A shared secret required on /ws and /api; None (the default) leaves both open, which is what a loopback-bound development server wants.	None
allowed_origins	frozenset of str	Browser origins accepted in addition to the request's own origin, e.g. http://localhost:5173 for a Vite dev server on another port. The single entry "*" accepts any origin.	frozenset()

token_required property

token_required: bool

Whether a token must be supplied to reach /ws and /api.

build classmethod

build(token: str | None, allowed_origins: Iterable[str]) -> WebSecurity

Build a policy from loose CLI/environment values.

An empty token string means "no token", which is what an unset environment variable and an unpassed option both arrive as.

Parameters:

Name	Type	Description	Default
token	str or None	The shared secret, or None/"" for none.	required
allowed_origins	Iterable of str	Extra origins to accept; blank entries are ignored.	required

Returns:

Name	Type	Description
security	WebSecurity	The configured policy.

Source code in src/indikit/web/security.py

@classmethod
def build(cls, token: str | None, allowed_origins: Iterable[str]) -> WebSecurity:
    """Build a policy from loose CLI/environment values.

    An empty token string means "no token", which is what an unset
    environment variable and an unpassed option both arrive as.

    Parameters
    ----------
    token : str or None
        The shared secret, or `None`/``""`` for none.
    allowed_origins : Iterable of str
        Extra origins to accept; blank entries are ignored.

    Returns
    -------
    security : WebSecurity
        The configured policy.
    """
    origins = frozenset(origin.strip() for origin in allowed_origins if origin.strip())
    return cls(token=token or None, allowed_origins=origins)

origin_allowed

origin_allowed(origin: str | None, host: str | None) -> bool

Report whether a handshake's Origin may open a connection.

Parameters:

Name	Type	Description	Default
origin	str or None	The request's Origin header; None when it carried none, which is allowed (see the module docstring).	required
host	str or None	The request's Host header, against which same-origin is judged.	required

Returns:

Name	Type	Description
allowed	bool	True when the connection may proceed.

Source code in src/indikit/web/security.py

def origin_allowed(self, origin: str | None, host: str | None) -> bool:
    """Report whether a handshake's ``Origin`` may open a connection.

    Parameters
    ----------
    origin : str or None
        The request's ``Origin`` header; `None` when it carried none, which
        is allowed (see the module docstring).
    host : str or None
        The request's ``Host`` header, against which same-origin is judged.

    Returns
    -------
    allowed : bool
        `True` when the connection may proceed.
    """
    if origin is None:
        return True
    if "*" in self.allowed_origins:
        return True
    if origin in self.allowed_origins:
        return True
    if host is None:
        return False
    # netloc, not the whole URL: an origin is scheme + host + port, and Host
    # carries host + port, so this is the only comparable pair. The port is
    # part of it - http://localhost:9999 is a different origin from :8000
    # even though a cookie would not distinguish them.
    return urlsplit(origin).netloc.casefold() == host.casefold()

token_ok

token_ok(supplied: str | None) -> bool

Report whether a supplied token matches the configured one.

Parameters:

Name	Type	Description	Default
supplied	str or None	The token read from the request, or None if it carried none.	required

Returns:

Name	Type	Description
ok	bool	True when no token is configured, or when the supplied one matches.

Source code in src/indikit/web/security.py

def token_ok(self, supplied: str | None) -> bool:
    """Report whether a supplied token matches the configured one.

    Parameters
    ----------
    supplied : str or None
        The token read from the request, or `None` if it carried none.

    Returns
    -------
    ok : bool
        `True` when no token is configured, or when the supplied one matches.
    """
    if self.token is None:
        return True
    if supplied is None:
        return False
    return hmac.compare_digest(supplied, self.token)

is_loopback

is_loopback(host: str) -> bool

Report whether a bind address reaches only this machine.

Used by the CLI to refuse a network-facing bind that has no token on it. An empty host is not loopback: to a socket API it means "every interface", which is the exposure this check exists to catch.

Parameters:

Name	Type	Description	Default
host	str	A bind address as passed to --host: a hostname or an IP literal.	required

Returns:

Name	Type	Description
loopback	bool	True when the address is localhost or a loopback IP literal.

Source code in src/indikit/web/security.py

def is_loopback(host: str) -> bool:
    """Report whether a bind address reaches only this machine.

    Used by the CLI to refuse a network-facing bind that has no token on it. An
    empty host is **not** loopback: to a socket API it means "every interface",
    which is the exposure this check exists to catch.

    Parameters
    ----------
    host : str
        A bind address as passed to ``--host``: a hostname or an IP literal.

    Returns
    -------
    loopback : bool
        `True` when the address is ``localhost`` or a loopback IP literal.
    """
    if host.lower() in {"localhost", "localhost.localdomain"}:
        return True
    try:
        return ipaddress.ip_address(host.strip("[]")).is_loopback
    except ValueError:
        # A name that is not a literal could resolve anywhere, so treat it as
        # exposed: the wrong answer here is the one that stays quiet.
        return False