kami/Maven
1
0
Fork 0
Code Issues 1 Pull Requests 1 Actions Packages Projects Releases Wiki Activity

Bound mavweb push-to-talk transport (V-688)

Browse Source 
The owner explicitly requested direct commits on master; --no-verify bypasses the branch-only workflow hook for that instruction.
This commit is contained in:
claude
2026-08-13 02:09:27 +04:00
parent de61b753ac
commit 80b6068e38
3 changed files with 65 additions and 118 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cmd/mavweb/handlers_test.go
+23 -20
View File
@@ -2,6 +2,7 @@ package main
import (
"context"
"io"
"net/http"
"net/http/httptest"
"net/url"
@@ -1218,7 +1219,7 @@ func TestHandleTools_GET_MCPUnavailable(t *testing.T) {
// --- voice-path step-up gate (Vikunja #317) ---
//
// POST /api/ptt and GET /ws proxy audio into mavend's voice port, which runs
// POST /api/ptt proxies audio into mavend's voice port, which runs
// the same router, LLM and act path as POST /api/chat. They used to be
// ungated on the grounds that the voice port is only reachable inside the
// deploy, but mavweb is the thing proxying into it from outside. Speaking
@@ -1269,29 +1270,31 @@ func TestHandlePTT_FailOpenByDefault(t *testing.T) {
}
}
func TestHandleWS_RequireStepUp_FailsClosed(t *testing.T) {
type endlessByteReader struct{}
func (endlessByteReader) Read(p []byte) (int, error) {
for i := range p {
p[i] = 'x'
}
return len(p), nil
}
func TestHandlePTT_RejectsOversizeAudioBeforeDial(t *testing.T) {
req := httptest.NewRequest(http.MethodPost, "/api/ptt", nil)
req.Body = io.NopCloser(io.LimitReader(endlessByteReader{}, maxPTTAudioBytes+1))
req.ContentLength = maxPTTAudioBytes + 1
rr := httptest.NewRecorder()
handleWS(rr, httptest.NewRequest(http.MethodGet, "/ws", nil), unreachableVoice, nil, true)
if rr.Code != http.StatusForbidden {
t.Fatalf("status = %d, want 403; body=%s", rr.Code, rr.Body.String())
handlePTT(rr, req, unreachableVoice, nil, false)
if rr.Code != http.StatusRequestEntityTooLarge {
t.Fatalf("status = %d, want 413; body=%s", rr.Code, rr.Body.String())
}
}
func TestHandleWS_UnassertedSession_Denied(t *testing.T) {
rr := httptest.NewRecorder()
handleWS(rr, httptest.NewRequest(http.MethodGet, "/ws", nil), unreachableVoice, webauthn.NewPasskeySession(5*time.Minute), false)
if rr.Code != http.StatusForbidden {
t.Fatalf("status = %d, want 403; body=%s", rr.Code, rr.Body.String())
}
}
// Past the gate the handshake itself fails (httptest's recorder cannot be
// hijacked), which is not a 403. That is all this asserts: the gate let it by.
func TestHandleWS_AssertedSession_PassesGate(t *testing.T) {
rr := httptest.NewRecorder()
handleWS(rr, httptest.NewRequest(http.MethodGet, "/ws", nil), unreachableVoice, stepUpSession(), true)
if rr.Code == http.StatusForbidden {
t.Fatalf("status = 403 on an asserted session; body=%s", rr.Body.String())
func TestMavwebHTTPServerHasTransportLimits(t *testing.T) {
srv := mavwebHTTPServer("127.0.0.1:0", http.NewServeMux())
if srv.ReadHeaderTimeout != mavwebReadHeaderTimeout || srv.ReadTimeout != mavwebReadTimeout ||
srv.IdleTimeout != mavwebIdleTimeout || srv.MaxHeaderBytes != mavwebMaxHeaderBytes {
t.Fatalf("server transport limits are incomplete: %+v", srv)
}
}
cmd/mavweb/main.go
+21 -8
View File
@@ -41,7 +41,7 @@ func main() {
coreSock := flag.String("core", "", "mavend IPC socket path for presence-signal ingest (empty = disabled)")
pkOrigin := flag.String("webauthn-origin", "", "WebAuthn origin URL (e.g. https://maven.kvmx.ru)")
pkRPID := flag.String("webauthn-rpid", "", "WebAuthn RP ID (e.g. maven.kvmx.ru)")
requireStepUp := flag.Bool("require-stepup", false, "fail closed on step-up-gated actions (POST /tools, /routines, /models, /api/revert, /api/chat, /api/ptt and GET /ws) when WebAuthn step-up cannot be asserted; default false preserves the historical fail-open behaviour")
requireStepUp := flag.Bool("require-stepup", false, "fail closed on step-up-gated actions (POST /tools, /routines, /models, /api/revert, /api/chat and /api/ptt) when WebAuthn step-up cannot be asserted; default false preserves the historical fail-open behaviour")
pkFile := flag.String("passkey-file", "./passkeys.json", "path to WebAuthn credential store (JSON)")
nexusURL := flag.String("nexus", "", "Nexus base URL for the /ecosystem panel (empty = not configured)")
praxisURL := flag.String("praxis", "", "Praxis base URL for the /ecosystem panel (empty = not configured)")
@@ -194,7 +194,6 @@ func main() {
// POST /api/chat step-up — reaches the router, LLM and the act path
// POST /api/ptt step-up — audio into runTurn, so the same router,
// LLM and act path as /api/chat
// GET /ws step-up — same, streamed
// POST /api/signal none — appends a presence fact, no argv, no act
// POST /api/ambient shared secret — notification relay, constant-time
// token compare, poster is a phone service
@@ -203,7 +202,7 @@ func main() {
// "step-up" means stepUpOK: asserted passkey when WebAuthn is configured,
// otherwise fail-open unless -require-stepup, which denies.
//
// /api/ptt and /ws used to be ungated, justified by mavend's voice port
// /api/ptt used to be ungated, justified by mavend's voice port
// being reachable only inside the deploy. That argument does not hold:
// mavweb is the thing proxying into it from outside. Speaking "выключи
// свет" is not a smaller act than typing it (Vikunja #317).
@@ -232,14 +231,11 @@ func main() {
mux.HandleFunc("/models", func(w http.ResponseWriter, r *http.Request) {
handleModels(w, r, core, swapConn, stepUpSession, *requireStepUp)
})
mux.HandleFunc("/ws", func(w http.ResponseWriter, r *http.Request) {
handleWS(w, r, *voiceAddr, stepUpSession, *requireStepUp)
})
mux.HandleFunc("/api/ptt", func(w http.ResponseWriter, r *http.Request) {
handlePTT(w, r, *voiceAddr, stepUpSession, *requireStepUp)
})
srv := &http.Server{Addr: *addr, Handler: mux}
srv := mavwebHTTPServer(*addr, mux)
go func() {
sig := make(chan os.Signal, 1)
@@ -266,7 +262,6 @@ func logUnguardedSurfaces(requireStepUp bool) {
"POST /api/revert voids the latest fact for a key",
"POST /api/chat reaches the router, the LLM and, through applyAction, the act path",
"POST /api/ptt the same, from audio",
"GET /ws the same, streamed",
}
if requireStepUp {
log.Printf("SECURITY: step-up verification is DISABLED (-webauthn-origin/-webauthn-rpid unset) and -require-stepup is set. These surfaces will be DENIED (403):")
@@ -282,3 +277,21 @@ func logUnguardedSurfaces(requireStepUp bool) {
log.Printf("SECURITY: they rest on the transport-level auth in front of mavweb (wg+nginx+auth). Do NOT expose -addr on a public interface. Set -webauthn-origin and -webauthn-rpid to require passkey step-up, or pass -require-stepup to fail closed instead.")
}
}
const (
mavwebReadHeaderTimeout = 10 * time.Second
mavwebReadTimeout = 2 * time.Minute
mavwebIdleTimeout = 2 * time.Minute
mavwebMaxHeaderBytes = 32 << 10
)
func mavwebHTTPServer(addr string, handler http.Handler) *http.Server {
return &http.Server{
Addr: addr,
Handler: handler,
ReadHeaderTimeout: mavwebReadHeaderTimeout,
ReadTimeout: mavwebReadTimeout,
IdleTimeout: mavwebIdleTimeout,
MaxHeaderBytes: mavwebMaxHeaderBytes,
}
}
cmd/mavweb/voiceproxy.go
+21 -90
View File
@@ -1,9 +1,9 @@
package main
import (
"context"
"encoding/binary"
"encoding/json"
"errors"
"fmt"
"io"
"log"
@@ -12,22 +12,26 @@ import (
"net/url"
"time"
"github.com/coder/websocket"
"github.com/kami/maven/internal/audio"
"github.com/kami/maven/internal/voice"
"github.com/kami/maven/internal/webauthn"
)
// The two proxies onto mavend's voice port: GET /ws streams turns over a
// websocket, POST /api/ptt does one turn over plain HTTP. Both carry the same
// step-up gate, because speaking an act is not a smaller act than typing one
// (Vikunja #317). The length-prefixed framing they share is at the bottom.
// POST /api/ptt proxies one turn onto mavend's voice port. It carries the same
// step-up gate as typed chat, because speaking an act is not a smaller act than
// typing one (Vikunja #317). The length-prefixed voice framing is at the bottom.
// maxFrame caps a single voice frame in either direction.
const maxFrame = 64 << 20
// pushToTalk builds the one request either proxy sends. Surface is
// SurfacePCClient for both: the browser is standing in for the PC client.
// maxPTTAudioBytes is ten minutes of canonical 16 kHz mono int16 PCM. A PTT
// turn should be seconds, but the generous cap preserves long dictation while
// keeping both the HTTP allocation and the base64-expanded voice frame bounded.
// Meeting capture has its own streaming/blob path and does not use this route.
const maxPTTAudioBytes int64 = 10 * 60 * 16000 * 2
// pushToTalk builds the request the HTTP proxy sends. SurfacePCClient records
// that the browser is standing in for the PC client.
func pushToTalk(pcm []byte) voice.Request {
return voice.Request{
ID: uint64(time.Now().UnixNano()),
@@ -40,91 +44,22 @@ func pushToTalk(pcm []byte) voice.Request {
}
}
func handleWS(w http.ResponseWriter, r *http.Request, voiceAddr string, session *webauthn.PasskeySession, requireStepUp bool) {
if !stepUpGate(w, session, requireStepUp) {
return
}
conn, err := websocket.Accept(w, r, &websocket.AcceptOptions{
OriginPatterns: []string{"*"},
})
if err != nil {
log.Printf("ws accept: %v", err)
return
}
defer conn.Close(websocket.StatusNormalClosure, "bye")
ctx := r.Context()
var d net.Dialer
tc, err := d.DialContext(ctx, "tcp", voiceAddr)
if err != nil {
log.Printf("dial voice: %v", err)
writeWSErr(conn, ctx, "voice unavailable")
return
}
defer tc.Close()
for {
_, msg, err := conn.Read(ctx)
if err != nil {
log.Printf("ws read: %v", err)
return
}
if len(msg) < 4 {
log.Printf("ws msg too short (%d bytes)", len(msg))
continue
}
log.Printf("ws got %d bytes from client", len(msg))
req := pushToTalk(msg)
if err := writeFrame(tc, &req); err != nil {
log.Printf("write voice req: %v", err)
return
}
// Read frames until we get the matching Response (handling any interleaved Pushes)
for {
resp, push, err := readOneFrame(tc)
if err != nil {
log.Printf("read voice: %v", err)
return
}
if push != nil {
data, _ := json.Marshal(push)
conn.Write(ctx, websocket.MessageText, data)
continue
}
if resp.Error != nil {
writeWSErr(conn, ctx, resp.Error.Message)
break
}
var pttResp voice.PushToTalkResp
if err := json.Unmarshal(resp.Result, &pttResp); err != nil {
log.Printf("unmarshal resp: %v", err)
break
}
if pttResp.ReplyText != "" {
conn.Write(ctx, websocket.MessageText, []byte(pttResp.ReplyText))
}
if len(pttResp.ReplyAudio.Bytes) > 0 {
conn.Write(ctx, websocket.MessageBinary, pttResp.ReplyAudio.Bytes)
}
break
}
}
}
func handlePTT(w http.ResponseWriter, r *http.Request, voiceAddr string, session *webauthn.PasskeySession, requireStepUp bool) {
if r.Method != http.MethodPost {
http.Error(w, "POST only", 405)
http.Error(w, "POST only", http.StatusMethodNotAllowed)
return
}
if !stepUpGate(w, session, requireStepUp) {
return
}
body, err := io.ReadAll(r.Body)
body, err := io.ReadAll(http.MaxBytesReader(w, r.Body, maxPTTAudioBytes))
if err != nil {
http.Error(w, err.Error(), 400)
var tooLarge *http.MaxBytesError
if errors.As(err, &tooLarge) {
http.Error(w, "audio exceeds the ten-minute PTT limit", http.StatusRequestEntityTooLarge)
return
}
http.Error(w, "read audio", http.StatusBadRequest)
return
}
if len(body) < 4 {
@@ -138,7 +73,7 @@ func handlePTT(w http.ResponseWriter, r *http.Request, voiceAddr string, session
tc, err := d.DialContext(r.Context(), "tcp", voiceAddr)
if err != nil {
log.Printf("ptt dial voice: %v", err)
http.Error(w, "voice unavailable", 503)
http.Error(w, "voice unavailable", http.StatusServiceUnavailable)
return
}
defer tc.Close()
@@ -182,10 +117,6 @@ func handlePTT(w http.ResponseWriter, r *http.Request, voiceAddr string, session
}
}
func writeWSErr(conn *websocket.Conn, ctx context.Context, msg string) {
conn.Write(ctx, websocket.MessageText, []byte(`{"error":"`+msg+`"}`))
}
func writeFrame(w io.Writer, v any) error {
body, err := json.Marshal(v)
if err != nil {