Behavision/desktop/stream_proxy_test.go

package main

import (
	"fmt"
	"io"
	"net/http"
	"net/http/httptest"
	"os"
	"strings"
	"testing"
	"time"
)

// fakeEngine stands in for the Python engine: it demands Basic auth exactly as
// the real one does when a credential is configured, and records what it was
// asked for.
type fakeEngine struct {
	*httptest.Server
	gotPath string
	gotUser string
	gotPass string
	hadAuth bool
}

func newFakeEngine(t *testing.T, body string) *fakeEngine {
	t.Helper()
	f := &fakeEngine{}
	f.Server = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		f.gotPath = r.URL.Path
		if r.URL.RawQuery != "" {
			f.gotPath += "?" + r.URL.RawQuery
		}
		f.gotUser, f.gotPass, f.hadAuth = r.BasicAuth()
		if !f.hadAuth {
			w.Header().Set("WWW-Authenticate", `Basic realm="behavision"`)
			w.WriteHeader(http.StatusUnauthorized)
			return
		}
		w.Header().Set("Content-Type", "multipart/x-mixed-replace; boundary=frame")
		_, _ = io.WriteString(w, body)
	}))
	t.Cleanup(f.Close)
	return f
}

func startProxy(t *testing.T, engine string, user, pass string) *streamProxy {
	t.Helper()
	p := newStreamProxy()
	if err := p.start(engine, user, pass); err != nil {
		t.Fatalf("start: %v", err)
	}
	t.Cleanup(p.stop)
	return p
}

func get(t *testing.T, url string) (int, string) {
	t.Helper()
	c := &http.Client{Timeout: 5 * time.Second}
	resp, err := c.Get(url)
	if err != nil {
		t.Fatalf("get %s: %v", url, err)
	}
	defer resp.Body.Close()
	b, _ := io.ReadAll(resp.Body)
	return resp.StatusCode, string(b)
}

// The whole point: the webview gets a URL it can actually load, and the
// password stays behind. A credential in the src is both unloadable in a
// Chromium webview and readable by anything that can see the DOM.
func TestTheCameraURLCarriesNoPassword(t *testing.T) {
	engine := newFakeEngine(t, "frames")
	p := startProxy(t, engine.URL, "behavision", "hunter2-the-real-one")

	u := p.urlFor("cam2", "stream.mjpeg")
	if u == "" {
		t.Fatal("no url while the proxy is running")
	}
	if strings.Contains(u, "hunter2-the-real-one") || strings.Contains(u, "behavision:") {
		t.Fatalf("credential leaked into the tile URL: %s", u)
	}
	if !strings.HasPrefix(u, "http://127.0.0.1:") {
		t.Fatalf("relay must be loopback only, got %s", u)
	}
}

func TestTheRelayAttachesTheCredentialItself(t *testing.T) {
	engine := newFakeEngine(t, "frame-bytes")
	p := startProxy(t, engine.URL, "behavision", "s3cret")

	code, body := get(t, p.urlFor("cam2", "stream.mjpeg"))
	if code != http.StatusOK {
		t.Fatalf("want 200 through the relay, got %d", code)
	}
	if body != "frame-bytes" {
		t.Fatalf("body not relayed unchanged: %q", body)
	}
	if !engine.hadAuth || engine.gotUser != "behavision" || engine.gotPass != "s3cret" {
		t.Fatalf("engine did not receive the credential: auth=%v user=%q",
			engine.hadAuth, engine.gotUser)
	}
	if engine.gotPath != "/api/cameras/cam2/stream.mjpeg" {
		t.Fatalf("wrong upstream path: %s", engine.gotPath)
	}
}

// The token is what keeps every other process on a shop PC from opening a live
// view of customers' faces, now that the relay itself has no password.
func TestAnotherProcessCannotGuessItsWayIn(t *testing.T) {
	engine := newFakeEngine(t, "frames")
	p := startProxy(t, engine.URL, "behavision", "s3cret")
	addr := p.ln.Addr().String()

	for _, bad := range []string{"", "0", strings.Repeat("a", 64), "wrong-token"} {
		url := fmt.Sprintf("http://%s/s/%s/cam2/stream.mjpeg", addr, bad)
		if code, _ := get(t, url); code != http.StatusNotFound {
			t.Fatalf("token %q got %d, want 404", bad, code)
		}
	}
	if engine.hadAuth {
		t.Fatal("a rejected request still reached the engine")
	}
}

// A camera id is interpolated into the upstream path, so it has to be a camera
// id and not a way to walk to a different endpoint with the credential
// attached.
func TestACameraIdCannotClimbOutOfItsPath(t *testing.T) {
	engine := newFakeEngine(t, "frames")
	p := startProxy(t, engine.URL, "behavision", "s3cret")
	addr := p.ln.Addr().String()

	for _, bad := range []string{"..", "%2e%2e", "cam2/../../api/identities", "cam 2", ""} {
		url := fmt.Sprintf("http://%s/s/%s/%s/stream.mjpeg", addr, p.token, bad)
		code, _ := get(t, url)
		if code != http.StatusNotFound {
			t.Fatalf("camera id %q got %d, want 404", bad, code)
		}
	}
	if strings.Contains(engine.gotPath, "identities") {
		t.Fatalf("reached a non-camera endpoint: %s", engine.gotPath)
	}
}

// Only the two files a tile needs. The engine also serves the identity list and
// the erasure endpoint; holding the token must not open those.
func TestOnlyTheTwoCameraFilesAreReachable(t *testing.T) {
	engine := newFakeEngine(t, "frames")
	p := startProxy(t, engine.URL, "behavision", "s3cret")
	addr := p.ln.Addr().String()

	for _, bad := range []string{"identities", "stats", "commission", "stream.mjpeg.bak"} {
		url := fmt.Sprintf("http://%s/s/%s/cam2/%s", addr, p.token, bad)
		if code, _ := get(t, url); code != http.StatusNotFound {
			t.Fatalf("file %q got %d, want 404", bad, code)
		}
	}

	for _, good := range []string{"stream.mjpeg", "frame.jpg"} {
		url := fmt.Sprintf("http://%s/s/%s/cam2/%s", addr, p.token, good)
		if code, _ := get(t, url); code != http.StatusOK {
			t.Fatalf("file %q got %d, want 200", good, code)
		}
	}
}

// frame.jpg takes width and quality - the engine re-encodes on demand, and a
// relay that dropped the query would silently serve full-size frames.
func TestTheQueryStringSurvivesTheRelay(t *testing.T) {
	engine := newFakeEngine(t, "frames")
	p := startProxy(t, engine.URL, "behavision", "s3cret")

	url := p.urlFor("cam2", "frame.jpg") + "?width=640&quality=70"
	if code, _ := get(t, url); code != http.StatusOK {
		t.Fatalf("got %d", code)
	}
	if !strings.Contains(engine.gotPath, "width=640") ||
		!strings.Contains(engine.gotPath, "quality=70") {
		t.Fatalf("query dropped: %s", engine.gotPath)
	}
}

// An engine that is not running must read as a bad gateway, not as a hang. A
// blank tile that never resolves is the symptom this whole file exists to end.
func TestAnEngineThatIsDownFailsQuickly(t *testing.T) {
	// Port 1 on loopback: nothing listens, and the connection is refused
	// rather than dropped, so this is fast and deterministic.
	p := startProxy(t, "http://127.0.0.1:1", "behavision", "s3cret")

	done := make(chan int, 1)
	go func() { code, _ := get(t, p.urlFor("cam2", "stream.mjpeg")); done <- code }()
	select {
	case code := <-done:
		if code != http.StatusBadGateway {
			t.Fatalf("want 502, got %d", code)
		}
	case <-time.After(8 * time.Second):
		t.Fatal("a dead engine left the request hanging")
	}
}

// Stopping must actually free the port, or a restarted engine leaves listeners
// behind for the life of the process.
func TestStoppingReleasesEverything(t *testing.T) {
	engine := newFakeEngine(t, "frames")
	p := newStreamProxy()
	if err := p.start(engine.URL, "u", "p"); err != nil {
		t.Fatalf("start: %v", err)
	}
	url := p.urlFor("cam2", "stream.mjpeg")
	if code, _ := get(t, url); code != http.StatusOK {
		t.Fatalf("want 200 before stop, got %d", code)
	}

	p.stop()

	if got := p.urlFor("cam2", "stream.mjpeg"); got != "" {
		t.Fatalf("still handing out URLs after stop: %s", got)
	}
	c := &http.Client{Timeout: 3 * time.Second}
	if resp, err := c.Get(url); err == nil {
		resp.Body.Close()
		t.Fatal("listener still accepting after stop")
	}
}

// start twice must not leave two listeners, which is what a restarted engine
// would otherwise cause.
func TestStartingTwiceIsANoOp(t *testing.T) {
	engine := newFakeEngine(t, "frames")
	p := startProxy(t, engine.URL, "u", "p")

	first := p.urlFor("cam2", "stream.mjpeg")
	if err := p.start(engine.URL, "u", "p"); err != nil {
		t.Fatalf("second start: %v", err)
	}
	if second := p.urlFor("cam2", "stream.mjpeg"); second != first {
		t.Fatalf("second start moved the relay: %s -> %s", first, second)
	}
}

// Against the real engine, which the unit tests above deliberately do not
// touch. Skipped unless TEST_ENGINE_URL is set, the same rule the server's
// live store tests follow: the suite must stay runnable with no services.
//
//	TEST_ENGINE_URL=http://127.0.0.1:8010 \
//	TEST_ENGINE_USER=... TEST_ENGINE_PASS=... go test ./desktop/ -run Live
//
// It exists because everything above proves the relay against a fake that
// agrees with me. Only a real engine proves the thing that was actually
// broken: that a multipart MJPEG stream arrives through the relay in pieces,
// rather than being buffered into a tile that never paints.
func TestLiveRelayCarriesRealMJPEGFrames(t *testing.T) {
	base := os.Getenv("TEST_ENGINE_URL")
	if base == "" {
		t.Skip("set TEST_ENGINE_URL to run the live relay test")
	}
	cam := os.Getenv("TEST_ENGINE_CAMERA")
	if cam == "" {
		cam = "cam2"
	}
	p := startProxy(t, base, os.Getenv("TEST_ENGINE_USER"), os.Getenv("TEST_ENGINE_PASS"))

	url := p.urlFor(cam, "stream.mjpeg")
	req, _ := http.NewRequest(http.MethodGet, url, nil)
	resp, err := (&http.Client{}).Do(req)
	if err != nil {
		t.Fatalf("relay: %v", err)
	}
	defer resp.Body.Close()
	if resp.StatusCode != http.StatusOK {
		t.Fatalf("relay returned %d - the credential did not reach the engine", resp.StatusCode)
	}
	if ct := resp.Header.Get("Content-Type"); !strings.Contains(ct, "multipart") {
		t.Fatalf("not a stream: Content-Type %q", ct)
	}

	// Read until two JPEG start markers have gone past. One proves it opened;
	// two prove it is still delivering, which is the difference between a
	// working tile and a single frozen frame.
	deadline := time.Now().Add(15 * time.Second)
	var seen, total int
	buf := make([]byte, 16*1024)
	for seen < 2 && time.Now().Before(deadline) {
		n, rerr := resp.Body.Read(buf)
		total += n
		seen += strings.Count(string(buf[:n]), "\xff\xd8\xff")
		if rerr != nil {
			break
		}
	}
	if seen < 2 {
		t.Fatalf("only %d JPEG frames in %d bytes - the relay is not streaming", seen, total)
	}
	t.Logf("relayed %d frames in %d bytes with no credential in the URL", seen, total)
}