WebSocket Simulations

# pip install websockets
import asyncio, base64, os
from http import HTTPStatus
from websockets.asyncio.server import serve

EXPECTED = "Basic " + base64.b64encode(
    f"{os.environ['CHIRP_USER']}:{os.environ['CHIRP_PASS']}".encode()).decode()

def check_auth(connection, request):
    if request.headers.get("Authorization") != EXPECTED:
        return connection.respond(HTTPStatus.UNAUTHORIZED, "unauthorized\n")

async def handler(ws):
    async for msg in ws:
        if isinstance(msg, bytes):
            await ws.send(msg)  # replace the echo with your agent audio

async def main():
    async with serve(handler, "0.0.0.0", 8080, process_request=check_auth):
        await asyncio.Future()

asyncio.run(main())

// npm i ws
import { createServer } from 'http'
import { WebSocketServer } from 'ws'

const { CHIRP_USER, CHIRP_PASS } = process.env
const expected = 'Basic ' + Buffer.from(`${CHIRP_USER}:${CHIRP_PASS}`).toString('base64')

const server = createServer()
const wss = new WebSocketServer({ noServer: true })

server.on('upgrade', (req, socket, head) => {
  if (req.headers.authorization !== expected) {
    socket.write('HTTP/1.1 401 Unauthorized\r\n\r\n'); socket.destroy(); return
  }
  wss.handleUpgrade(req, socket, head, (ws) => wss.emit('connection', ws, req))
})

wss.on('connection', (ws) => {
  ws.on('message', (data, isBinary) => {
    if (isBinary) ws.send(data, { binary: true }) // replace the echo with your agent audio
  })
})

server.listen(8080)

package main

import (
    "crypto/subtle"
    "encoding/base64"
    "net/http"
    "os"

    "github.com/gorilla/websocket"
)

var up = websocket.Upgrader{CheckOrigin: func(r *http.Request) bool { return true }}

func main() {
    expected := "Basic " + base64.StdEncoding.EncodeToString(
        []byte(os.Getenv("CHIRP_USER")+":"+os.Getenv("CHIRP_PASS")))
    http.HandleFunc("/voice", func(w http.ResponseWriter, r *http.Request) {
        if subtle.ConstantTimeCompare([]byte(r.Header.Get("Authorization")), []byte(expected)) != 1 {
            http.Error(w, "unauthorized", 401); return
        }
        c, err := up.Upgrade(w, r, nil)
        if err != nil { return }
        defer c.Close()
        for {
            mt, msg, err := c.ReadMessage()
            if err != nil { return }
            if mt == websocket.BinaryMessage {
                _ = c.WriteMessage(websocket.BinaryMessage, msg)
            }
        }
    })
    http.ListenAndServe(":8080", nil)
}