funknetz/station/behavior_test.go

package station_test

import (
	"context"
	"crypto/hmac"
	"crypto/sha1"
	"encoding/binary"
	"io"
	"net"
	"strconv"
	"testing"
	"time"

	"gitea.neitzel.de/konrad/funknetz/internal/opus"
	"gitea.neitzel.de/konrad/funknetz/station"
)

type loggedIn struct {
	st     *station.Station
	clock  *fakeClock
	refl   *reflector
	cancel context.CancelFunc
	id     uint16
}

func login(t *testing.T, callsign, key string, tg int, id uint16, codecs []string) loggedIn {
	t.Helper()
	return loginLog(t, callsign, key, tg, id, codecs, "")
}

func loginLog(t *testing.T, callsign, key string, tg int, id uint16, codecs []string, logPath string) loggedIn {
	t.Helper()
	refl := newReflector(t)
	file := configJSON("127.0.0.1", refl.port, callsign, key, tg)
	if logPath != "" {
		file = configWithLog("127.0.0.1", refl.port, callsign, key, tg, logPath)
	}
	path := writeConfigFile(t, file)
	clock := newFakeClock()
	st, err := station.New(station.Options{ConfigPath: path, Clock: clock})
	if err != nil {
		t.Fatal(err)
	}
	ctx, cancel := context.WithCancel(t.Context())
	t.Cleanup(cancel)
	go func() { _ = st.Run(ctx) }()
	refl.accept(t)

	typ, body := refl.readFrame(t)
	if typ != 5 || len(body) != 4 || body[1] != 2 {
		t.Fatalf("protocol frame type %d body %v", typ, body)
	}
	challenge := make([]byte, 20)
	for i := range challenge {
		challenge[i] = byte(i + 1)
	}
	refl.writeFrame(t, 10, appendBytes(nil, challenge))
	typ, body = refl.readFrame(t)
	if typ != 11 {
		t.Fatalf("auth type %d", typ)
	}
	call, rest := takeStr(t, body)
	digest, _ := takeBytes(t, rest)
	if call != "DL1ABC" && call != callsign {
		t.Fatalf("callsign %q", call)
	}
	mac := hmac.New(sha1.New, []byte(key))
	mac.Write(challenge)
	if !hmac.Equal(digest, mac.Sum(nil)) {
		t.Fatal("auth digest does not match the stored key")
	}
	refl.writeFrame(t, 12, nil)
	refl.writeFrame(t, 100, serverInfo(id, []string{"db0abc"}, codecs))
	typ, body = refl.readFrame(t)
	if typ != 111 {
		t.Fatalf("type %d, want software identity", typ)
	}
	requireSoftwareIdentity(t, body)
	typ, body = refl.readFrame(t)
	if typ != 106 {
		t.Fatalf("type %d, want select talk group", typ)
	}
	gotTG, _ := takeU32(t, body)
	if int(gotTG) != tg {
		t.Fatalf("talk group %d, want %d", gotTG, tg)
	}
	utyp, gotID, seq, ubody := refl.readUDP(t)
	if utyp != 1 || gotID != id || seq != 0 || len(ubody) != 0 {
		t.Fatalf("udp heartbeat type %d id %d seq %d len %d", utyp, gotID, seq, len(ubody))
	}
	if _, err := st.Wait(ctx, func(s station.State) bool { return s.Connection == "Connected" }); err != nil {
		t.Fatal(err)
	}
	return loggedIn{st: st, clock: clock, refl: refl, cancel: cancel, id: id}
}

func TestQuietSessionSendsHeartbeats(t *testing.T) {
	in := login(t, "DL1ABC", "key", 20, 7, []string{"OPUS"})

	in.clock.BlockUntilWaiting(t)
	in.clock.Advance(9 * time.Second)
	if _, _, ok := tryFrame(in.refl); ok {
		t.Fatal("TCP heartbeat arrived before 10 seconds")
	}
	in.clock.Advance(1 * time.Second)
	in.clock.BlockUntilWaiting(t)
	typ, body := in.refl.readFrame(t)
	if typ != 1 || len(body) != 0 {
		t.Fatalf("TCP heartbeat type %d len %d", typ, len(body))
	}
	gen := in.clock.Generation()
	in.refl.writeFrame(t, 1, nil)
	waitClock(t, in.clock, gen)

	in.clock.Advance(5 * time.Second)
	utyp, id, seq, ubody := in.refl.readUDP(t)
	if utyp != 1 || id != 7 || seq != 1 || len(ubody) != 0 {
		t.Fatalf("UDP heartbeat type %d id %d seq %d len %d", utyp, id, seq, len(ubody))
	}
}

func TestDropRetriesUntilQuit(t *testing.T) {
	in := login(t, "DL1ABC", "key", 20, 7, []string{"OPUS"})
	in.refl.tcp.Close()
	if _, err := in.st.Wait(context.Background(), func(s station.State) bool {
		return s.Connection == "Disconnected" && s.Detail == "Idle"
	}); err != nil {
		t.Fatal(err)
	}
	in.clock.BlockUntilWaiting(t)
	in.clock.Advance(2 * time.Second)
	in.refl.accept(t)
	typ, body := in.refl.readFrame(t)
	if typ != 5 || body[1] != 2 {
		t.Fatalf("retry protocol type %d body %v", typ, body)
	}
	in.cancel()
	in.clock.Advance(2 * time.Second)
	in.refl.ln.(*net.TCPListener).SetDeadline(time.Now().Add(150 * time.Millisecond))
	if conn, err := in.refl.ln.Accept(); err == nil {
		conn.Close()
		t.Fatal("dialed after quit")
	}
}

func TestRefusedConnectionRetriesUntilQuit(t *testing.T) {
	ln, err := net.Listen("tcp", "127.0.0.1:0")
	if err != nil {
		t.Fatal(err)
	}
	port := ln.Addr().(*net.TCPAddr).Port
	ln.Close()

	path := writeConfigFile(t, configJSON("127.0.0.1", port, "DL1ABC", "key", 4))
	clock := newFakeClock()
	st, err := station.New(station.Options{ConfigPath: path, Clock: clock})
	if err != nil {
		t.Fatal(err)
	}
	ctx, cancel := context.WithCancel(t.Context())
	defer cancel()
	go func() { _ = st.Run(ctx) }()

	if _, err := st.Wait(ctx, func(s station.State) bool { return s.Connection == "Disconnected" }); err != nil {
		t.Fatal(err)
	}
	st.HoldButton()
	st.WriteMicrophone(make([]int16, opus.FrameSamples))

	ln, err = net.Listen("tcp", net.JoinHostPort("127.0.0.1", strconv.Itoa(port)))
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { ln.Close() })
	clock.BlockUntilWaiting(t)
	clock.Advance(2 * time.Second)
	ln.(*net.TCPListener).SetDeadline(time.Now().Add(2 * time.Second))
	conn, err := ln.Accept()
	if err != nil {
		t.Fatal(err)
	}
	conn.Close()
	if _, err := st.Wait(ctx, func(s station.State) bool { return s.Connection == "Disconnected" }); err != nil {
		t.Fatal(err)
	}
	cancel()
	clock.Advance(2 * time.Second)
	ln.(*net.TCPListener).SetDeadline(time.Now().Add(150 * time.Millisecond))
	if conn, err := ln.Accept(); err == nil {
		conn.Close()
		t.Fatal("dialed after quit")
	}
}

func TestReflectorErrorStopsRetries(t *testing.T) {
	for _, msg := range []string{"Access denied", "no such station"} {
		t.Run(msg, func(t *testing.T) {
			refl := newReflector(t)
			path := writeConfigFile(t, configJSON("127.0.0.1", refl.port, "DL1ABC", "key", 20))
			clock := newFakeClock()
			st, err := station.New(station.Options{ConfigPath: path, Clock: clock})
			if err != nil {
				t.Fatal(err)
			}
			ctx, cancel := context.WithCancel(t.Context())
			defer cancel()
			go func() { _ = st.Run(ctx) }()
			refl.accept(t)
			refl.readFrame(t)
			refl.writeFrame(t, 13, appendStr(nil, msg))
			got, err := st.Wait(ctx, func(s station.State) bool { return s.Detail == msg })
			if err != nil {
				t.Fatal(err)
			}
			if got.Connection != "Disconnected" {
				t.Fatalf("Connection = %s, want Disconnected", got.Connection)
			}
			clock.Advance(2 * time.Second)
			refl.ln.(*net.TCPListener).SetDeadline(time.Now().Add(150 * time.Millisecond))
			if conn, err := refl.ln.Accept(); err == nil {
				conn.Close()
				t.Fatal("retried after reflector error")
			}
			if got := st.State().Detail; got != msg {
				t.Fatalf("Detail = %q, want it to stay %q", got, msg)
			}
		})
	}
}

func TestMissingOpusStopsRetries(t *testing.T) {
	refl := newReflector(t)
	path := writeConfigFile(t, configJSON("127.0.0.1", refl.port, "DL1ABC", "key", 20))
	clock := newFakeClock()
	st, err := station.New(station.Options{ConfigPath: path, Clock: clock})
	if err != nil {
		t.Fatal(err)
	}
	ctx, cancel := context.WithCancel(t.Context())
	defer cancel()
	go func() { _ = st.Run(ctx) }()
	refl.accept(t)
	refl.readFrame(t)
	challenge := make([]byte, 20)
	refl.writeFrame(t, 10, appendBytes(nil, challenge))
	refl.readFrame(t)
	refl.writeFrame(t, 12, nil)
	refl.writeFrame(t, 100, serverInfo(9, nil, []string{"GSM"}))
	got, err := st.Wait(ctx, func(s station.State) bool {
		return s.Detail == "The reflector did not offer OPUS."
	})
	if err != nil {
		t.Fatal(err)
	}
	if got.Connection != "Disconnected" {
		t.Fatalf("Connection = %s", got.Connection)
	}
	clock.Advance(2 * time.Second)
	refl.ln.(*net.TCPListener).SetDeadline(time.Now().Add(150 * time.Millisecond))
	if conn, err := refl.ln.Accept(); err == nil {
		conn.Close()
		t.Fatal("retried after missing OPUS")
	}
}

func TestTCPReceiveTimeoutDropsTheSession(t *testing.T) {
	in := login(t, "DL1ABC", "key", 20, 7, []string{"OPUS"})
	in.clock.BlockUntilWaiting(t)
	in.clock.Advance(15 * time.Second)
	if _, err := in.st.Wait(context.Background(), func(s station.State) bool {
		return s.Connection == "Disconnected"
	}); err != nil {
		t.Fatal(err)
	}
	in.clock.BlockUntilWaiting(t)
	in.clock.Advance(2 * time.Second)
	in.refl.accept(t)
}

func TestUDPReceiveTimeoutDropsTheSession(t *testing.T) {
	in := login(t, "DL1ABC", "key", 20, 7, []string{"OPUS"})
	for step := 0; step < 12; step++ {
		in.clock.BlockUntilWaiting(t)
		in.clock.Advance(5 * time.Second)
		gen := in.clock.Generation()
		if replyTCP(t, in.refl) {
			waitClock(t, in.clock, gen)
		}
		discardUDP(t, in.refl)
		if in.st.State().Connection == "Disconnected" {
			if step < 11 {
				t.Fatalf("dropped on step %d", step)
			}
			return
		}
	}
	t.Fatal("session stayed up for 60 seconds without UDP")
}

func TestRemoteAudioAndTalker(t *testing.T) {
	in := login(t, "DL1ABC", "key", 20, 7, []string{"OPUS"})
	enc, err := opus.NewEncoder()
	if err != nil {
		t.Fatal(err)
	}
	defer enc.Close()
	pkt, err := enc.Encode(make([]int16, opus.FrameSamples))
	if err != nil {
		t.Fatal(err)
	}
	in.refl.writeUDP(t, 101, 1, 0, appendBytes(nil, pkt))
	in.refl.writeUDP(t, 101, 7, 1, appendBytes(nil, pkt))

	ctx := t.Context()
	pcm, err := in.st.WaitPlayback(ctx, opus.FrameSamples)
	if err != nil {
		t.Fatal(err)
	}
	if len(pcm) != opus.FrameSamples {
		t.Fatalf("playback %d samples", len(pcm))
	}
	if extra := in.st.TakePlayback(); len(extra) != 0 {
		t.Fatalf("played %d extra samples from the wrong client id", len(extra))
	}
	if got := in.st.State(); got.Talker != "" || got.Detail != "Idle" {
		t.Fatalf("State = %+v, want Idle before the talker is named", got)
	}

	in.refl.writeFrame(t, 104, appendStr(appendU32(nil, 99), "OTHER1"))
	waitCtx, cancel := context.WithTimeout(ctx, 100*time.Millisecond)
	if _, err := in.st.Wait(waitCtx, func(s station.State) bool { return s.Talker == "OTHER1" }); err == nil {
		cancel()
		t.Fatal("showed a talker from another talk group")
	}
	cancel()

	in.refl.writeFrame(t, 104, appendStr(appendU32(nil, 20), "DB0XYZ"))
	got, err := in.st.Wait(ctx, func(s station.State) bool { return s.Talker == "DB0XYZ" })
	if err != nil {
		t.Fatal(err)
	}
	if got.Detail != "Idle" {
		t.Fatalf("Detail = %q", got.Detail)
	}
	in.st.HoldButton()
	in.st.WriteMicrophone(make([]int16, opus.FrameSamples))
	if _, _, ok := tryUDP(in.refl); ok {
		t.Fatal("sent audio while the room was busy")
	}

	in.refl.writeFrame(t, 105, appendStr(appendU32(nil, 20), "DB0XYZ"))
	if _, err := in.st.Wait(ctx, func(s station.State) bool {
		return s.Talker == "DL1ABC" && s.Transmitting
	}); err != nil {
		t.Fatal(err)
	}
	in.st.WriteMicrophone(make([]int16, opus.FrameSamples))
	utyp, id, _, body := in.refl.readUDP(t)
	if utyp != 101 || id != 7 {
		t.Fatalf("audio type %d id %d", utyp, id)
	}
	packet, _ := takeBytes(t, body)
	dec, err := opus.NewDecoder()
	if err != nil {
		t.Fatal(err)
	}
	defer dec.Close()
	decoded, err := dec.Decode(packet)
	if err != nil {
		t.Fatal(err)
	}
	if len(decoded) != opus.FrameSamples {
		t.Fatalf("decoded %d samples", len(decoded))
	}
}

func TestPlaybackPausesWhenTransmitting(t *testing.T) {
	in := login(t, "DL1ABC", "key", 20, 7, []string{"OPUS"})
	enc, err := opus.NewEncoder()
	if err != nil {
		t.Fatal(err)
	}
	defer enc.Close()
	pkt, err := enc.Encode(make([]int16, opus.FrameSamples))
	if err != nil {
		t.Fatal(err)
	}
	in.refl.writeUDP(t, 101, 7, 1, appendBytes(nil, pkt))
	in.refl.writeUDP(t, 101, 7, 2, appendBytes(nil, pkt))
	if _, err := in.st.WaitPlayback(t.Context(), 2*opus.FrameSamples-1); err != nil {
		t.Fatal(err)
	}
	in.st.HoldButton()
	in.refl.writeFrame(t, 105, appendStr(appendU32(nil, 20), "DB0XYZ"))
	if _, err := in.st.Wait(t.Context(), func(s station.State) bool { return s.Transmitting }); err != nil {
		t.Fatal(err)
	}
	if pcm := in.st.TakePlayback(); len(pcm) != 0 {
		t.Fatalf("playback %d samples, want none while transmitting", len(pcm))
	}
}

func TestMicrophoneIsNotPlayed(t *testing.T) {
	in := login(t, "DL1ABC", "key", 20, 7, []string{"OPUS"})
	in.st.HoldButton()
	in.st.WriteMicrophone(make([]int16, opus.FrameSamples))
	in.refl.readUDP(t)
	if pcm := in.st.TakePlayback(); len(pcm) != 0 {
		t.Fatalf("playback %d samples from the microphone", len(pcm))
	}
	in.st.ReleaseButton()
	in.refl.readUDP(t) // flush, or padded frame then flush; drained below
	discardUDP(t, in.refl)
}

func TestPushToTalk(t *testing.T) {
	in := login(t, "DL1ABC", "key", 20, 7, []string{"OPUS"})
	dec, err := opus.NewDecoder()
	if err != nil {
		t.Fatal(err)
	}
	defer dec.Close()

	in.st.WriteMicrophone(make([]int16, opus.FrameSamples*10))
	if _, _, ok := tryUDP(in.refl); ok {
		t.Fatal("sent audio before push-to-talk")
	}

	in.st.HoldButton()
	in.st.HoldSpace()
	if got := in.st.State(); got.Talker != "DL1ABC" || !got.Transmitting {
		t.Fatalf("State = %+v, want this station as the talker", got)
	}
	in.st.WriteMicrophone(make([]int16, opus.FrameSamples))
	expectAudio(t, in.refl, dec, 7, 1)
	in.st.ReleaseButton()
	in.st.WriteMicrophone(make([]int16, opus.FrameSamples))
	expectAudio(t, in.refl, dec, 7, 2)

	in.st.WriteMicrophone(make([]int16, 100))
	in.st.ReleaseSpace()
	expectAudio(t, in.refl, dec, 7, 3)
	utyp, id, seq, body := in.refl.readUDP(t)
	if utyp != 102 || id != 7 || seq != 4 || len(body) != 0 {
		t.Fatalf("flush type %d id %d seq %d len %d", utyp, id, seq, len(body))
	}
	if got := in.st.State(); got.Transmitting || got.Talker != "" || got.Detail != "Idle" {
		t.Fatalf("State = %+v, want Idle after release", got)
	}
	in.st.WriteMicrophone(make([]int16, opus.FrameSamples))
	if _, _, ok := tryUDP(in.refl); ok {
		t.Fatal("sent audio after release")
	}

	enc, err := opus.NewEncoder()
	if err != nil {
		t.Fatal(err)
	}
	defer enc.Close()
	pkt, err := enc.Encode(make([]int16, opus.FrameSamples))
	if err != nil {
		t.Fatal(err)
	}
	in.st.HoldSpace()
	in.refl.writeUDP(t, 101, 7, 9, appendBytes(nil, pkt))
	waitCtx, cancel := context.WithTimeout(t.Context(), 150*time.Millisecond)
	defer cancel()
	if _, err := in.st.WaitPlayback(waitCtx, opus.FrameSamples); err == nil {
		t.Fatal("played remote audio while transmitting")
	}
}

func TestSequenceWraps(t *testing.T) {
	in := login(t, "DL1ABC", "key", 20, 7, []string{"OPUS"})
	in.st.HoldButton()
	var prev uint16 = 0
	for i := 1; i <= 65536; i++ {
		in.st.WriteMicrophone(make([]int16, opus.FrameSamples))
		_, _, seq, _ := in.refl.readUDP(t)
		want := uint16(i)
		if seq != want {
			t.Fatalf("packet %d sequence %d, want %d", i, seq, want)
		}
		if i > 1 && seq != prev+1 {
			t.Fatalf("sequence %d followed %d", seq, prev)
		}
		prev = seq
	}
}

func expectAudio(t *testing.T, refl *reflector, dec *opus.Decoder, id, seq uint16) {
	t.Helper()
	typ, gotID, gotSeq, body := refl.readUDP(t)
	if typ != 101 || gotID != id || gotSeq != seq {
		t.Fatalf("datagram type %d id %d seq %d, want audio id %d seq %d", typ, gotID, gotSeq, id, seq)
	}
	packet, rest := takeBytes(t, body)
	if len(rest) != 0 {
		t.Fatalf("audio has %d trailing bytes", len(rest))
	}
	pcm, err := dec.Decode(packet)
	if err != nil {
		t.Fatal(err)
	}
	if len(pcm) != opus.FrameSamples {
		t.Fatalf("decoded %d samples", len(pcm))
	}
}

func tryFrame(r *reflector) (uint16, []byte, bool) {
	_ = r.tcp.SetReadDeadline(time.Now().Add(30 * time.Millisecond))
	var hdr [4]byte
	if _, err := io.ReadFull(r.tcp, hdr[:]); err != nil {
		return 0, nil, false
	}
	n := binary.BigEndian.Uint32(hdr[:])
	buf := make([]byte, n)
	if _, err := io.ReadFull(r.tcp, buf); err != nil {
		return 0, nil, false
	}
	if len(buf) < 2 {
		return 0, nil, false
	}
	return binary.BigEndian.Uint16(buf[:2]), buf[2:], true
}

func tryUDP(r *reflector) ([]byte, *net.UDPAddr, bool) {
	_ = r.udp.SetReadDeadline(time.Now().Add(40 * time.Millisecond))
	buf := make([]byte, 2048)
	n, addr, err := r.udp.ReadFromUDP(buf)
	if err != nil {
		return nil, nil, false
	}
	r.peer = addr
	return append([]byte(nil), buf[:n]...), addr, true
}

func replyTCP(t *testing.T, r *reflector) bool {
	t.Helper()
	wrote := false
	for {
		typ, _, ok := tryFrame(r)
		if !ok {
			return wrote
		}
		if typ == 1 {
			r.writeFrame(t, 1, nil)
			wrote = true
		}
	}
}

func discardUDP(t *testing.T, r *reflector) {
	t.Helper()
	for {
		if _, _, ok := tryUDP(r); !ok {
			return
		}
	}
}