Join one FM-Funknetz talk group from a headset window.

The station reads ~/.funknetz.json, speaks protocol 2.0, and builds an unsigned funknetz.app.

Co-authored-by: Cursor <cursoragent@cursor.com>

station/behavior_test.go

@@ -0,0 +1,520 @@
+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()
+	refl := newReflector(t)
+	path := writeConfigFile(t, configJSON("127.0.0.1", refl.port, callsign, key, tg))
+	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 != 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 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
+		}
+	}
+}

station/clock.go

@@ -0,0 +1,32 @@
+package station
+
+import "time"
+
+// Clock is the time source for heartbeats and the retry wait.
+type Clock interface {
+	Now() time.Time
+	NewTimer(d time.Duration) Timer
+}
+
+// Timer is a single deadline. Stop prevents the channel from firing.
+type Timer interface {
+	C() <-chan time.Time
+	Stop() bool
+}
+
+type wallClock struct{}
+
+func (wallClock) Now() time.Time { return time.Now() }
+
+func (wallClock) NewTimer(d time.Duration) Timer {
+	if d < 0 {
+		d = 0
+	}
+	return wallTimer{t: time.NewTimer(d)}
+}
+
+type wallTimer struct{ t *time.Timer }
+
+func (w wallTimer) C() <-chan time.Time { return w.t.C }
+
+func (w wallTimer) Stop() bool { return w.t.Stop() }

station/config.go

@@ -0,0 +1,155 @@
+package station
+
+import (
+	"fmt"
+	"math"
+	"os"
+	"regexp"
+	"strings"
+
+	"encoding/json/jsontext"
+	jsonv2 "encoding/json/v2"
+)
+
+type loaded struct {
+	host      string
+	port      int
+	callsign  string
+	authKey   string
+	talkGroup int
+}
+
+type fileShape struct {
+	Host      jsontext.Value `json:"host"`
+	Port      jsontext.Value `json:"port"`
+	Callsign  jsontext.Value `json:"callsign"`
+	AuthKey   jsontext.Value `json:"authKey"`
+	TalkGroup jsontext.Value `json:"talkGroup"`
+}
+
+type templateFile struct {
+	Host      string `json:"host"`
+	Port      int    `json:"port"`
+	Callsign  string `json:"callsign"`
+	AuthKey   string `json:"authKey"`
+	TalkGroup int    `json:"talkGroup"`
+}
+
+var callsignPattern = regexp.MustCompile(`^[A-Z0-9][A-Z]{0,2}\d[A-Z0-9]{0,3}[A-Z](?:-[A-Z0-9]{1,3})?$`)
+
+func loadConfig(path string) (loaded, error) {
+	body, err := os.ReadFile(path)
+	if err != nil {
+		if os.IsNotExist(err) {
+			if err := writeTemplate(path); err != nil {
+				return loaded{}, err
+			}
+			return loaded{}, fmt.Errorf("created %s", path)
+		}
+		return loaded{}, fmt.Errorf("could not be read")
+	}
+	return parseConfig(body)
+}
+
+func writeTemplate(path string) error {
+	body, err := jsonv2.Marshal(templateFile{})
+	if err != nil {
+		return err
+	}
+	body = append(body, '\n')
+	f, err := os.OpenFile(path, os.O_WRONLY|os.O_CREATE|os.O_EXCL, 0o600)
+	if err != nil {
+		return err
+	}
+	_, werr := f.Write(body)
+	cerr := f.Close()
+	if werr != nil {
+		return werr
+	}
+	if cerr != nil {
+		return cerr
+	}
+	return os.Chmod(path, 0o600)
+}
+
+func parseConfig(body []byte) (loaded, error) {
+	var shape fileShape
+	if err := jsonv2.Unmarshal(body, &shape); err != nil {
+		return loaded{}, fmt.Errorf("could not be read")
+	}
+	host, err := stringField("host", shape.Host)
+	if err != nil {
+		return loaded{}, err
+	}
+	port, err := intField("port", shape.Port, 1, 65535)
+	if err != nil {
+		return loaded{}, err
+	}
+	call, err := stringField("callsign", shape.Callsign)
+	if err != nil {
+		return loaded{}, err
+	}
+	key, err := authField(shape.AuthKey)
+	if err != nil {
+		return loaded{}, err
+	}
+	tg, err := intField("talkGroup", shape.TalkGroup, 1, int(math.MaxUint32))
+	if err != nil {
+		return loaded{}, err
+	}
+
+	host = strings.TrimSpace(host)
+	if host == "" {
+		return loaded{}, fmt.Errorf("host is empty")
+	}
+	call = strings.ToUpper(strings.TrimSpace(call))
+	if call == "" {
+		return loaded{}, fmt.Errorf("callsign is empty")
+	}
+	if !callsignPattern.MatchString(call) {
+		return loaded{}, fmt.Errorf("callsign is invalid")
+	}
+	return loaded{host: host, port: port, callsign: call, authKey: key, talkGroup: tg}, nil
+}
+
+func stringField(name string, v jsontext.Value) (string, error) {
+	if len(v) == 0 {
+		return "", nil
+	}
+	if v.Kind() != jsontext.KindString {
+		return "", fmt.Errorf("%s has the wrong type", name)
+	}
+	var s string
+	if err := jsonv2.Unmarshal(v, &s); err != nil {
+		return "", fmt.Errorf("%s has the wrong type", name)
+	}
+	return s, nil
+}
+
+func authField(v jsontext.Value) (string, error) {
+	s, err := stringField("authKey", v)
+	if err != nil {
+		return "", err
+	}
+	if strings.TrimSpace(s) == "" {
+		return "", fmt.Errorf("authKey is empty")
+	}
+	return s, nil
+}
+
+func intField(name string, v jsontext.Value, min, max int) (int, error) {
+	if len(v) == 0 {
+		return 0, fmt.Errorf("%s is invalid", name)
+	}
+	if v.Kind() != jsontext.KindNumber {
+		return 0, fmt.Errorf("%s has the wrong type", name)
+	}
+	var n int
+	if err := jsonv2.Unmarshal(v, &n); err != nil {
+		return 0, fmt.Errorf("%s is invalid", name)
+	}
+	if n < min || n > max {
+		return 0, fmt.Errorf("%s is invalid", name)
+	}
+	return n, nil
+}

station/config_test.go

@@ -0,0 +1,178 @@
+package station_test
+
+import (
+	"os"
+	"path/filepath"
+	"strings"
+	"testing"
+
+	jsonv2 "encoding/json/v2"
+
+	"gitea.neitzel.de/konrad/funknetz/station"
+)
+
+func TestMissingConfigIsCreatedAndRejected(t *testing.T) {
+	path := filepath.Join(t.TempDir(), ".funknetz.json")
+
+	_, err := station.New(station.Options{ConfigPath: path})
+	if err == nil {
+		t.Fatal("New() error = nil, want a rejection after creating the file")
+	}
+	if !strings.Contains(err.Error(), "created") {
+		t.Fatalf("New() error = %q, want it to say the file was created", err.Error())
+	}
+
+	info, err := os.Stat(path)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if got, want := info.Mode().Perm(), os.FileMode(0o600); got != want {
+		t.Fatalf("mode = %o, want %o", got, want)
+	}
+
+	body, err := os.ReadFile(path)
+	if err != nil {
+		t.Fatal(err)
+	}
+	var got map[string]any
+	if err := jsonv2.Unmarshal(body, &got); err != nil {
+		t.Fatalf("created file is not JSON: %v\n%s", err, body)
+	}
+	if got["host"] != "" || got["callsign"] != "" || got["authKey"] != "" {
+		t.Fatalf("string fields = %#v, want empty host, callsign, and authKey", got)
+	}
+	if got["port"] != float64(0) || got["talkGroup"] != float64(0) {
+		t.Fatalf("numeric fields = %#v, want port and talkGroup at zero", got)
+	}
+}
+
+func TestExistingConfigIsNotRewritten(t *testing.T) {
+	path := filepath.Join(t.TempDir(), "funknetz.json")
+	original := []byte("{\"host\":\"127.0.0.1\",\"port\":5308,\"callsign\":\"DL1ABC\",\"authKey\":\"a b\",\"talkGroup\":20}\n")
+	if err := os.WriteFile(path, original, 0o640); err != nil {
+		t.Fatal(err)
+	}
+
+	if _, err := station.New(station.Options{ConfigPath: path}); err != nil {
+		t.Fatal(err)
+	}
+	got, err := os.ReadFile(path)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if string(got) != string(original) {
+		t.Fatalf("file = %q, want %q", got, original)
+	}
+	info, err := os.Stat(path)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if got, want := info.Mode().Perm(), os.FileMode(0o640); got != want {
+		t.Fatalf("mode = %o, want %o", got, want)
+	}
+}
+
+func TestInvalidConfigNamesTheField(t *testing.T) {
+	tests := []struct {
+		name string
+		body string
+		want string
+	}{
+		{name: "empty host", body: configJSON("", 5308, "DL1ABC", "key", 1), want: "host is empty"},
+		{name: "blank host", body: configJSON("   ", 5308, "DL1ABC", "key", 1), want: "host is empty"},
+		{name: "host type", body: `{"host":1,"port":5308,"callsign":"DL1ABC","authKey":"key","talkGroup":1}`, want: "host has the wrong type"},
+		{name: "empty callsign", body: configJSON("127.0.0.1", 5308, "", "key", 1), want: "callsign is empty"},
+		{name: "callsign type", body: `{"host":"127.0.0.1","port":5308,"callsign":1,"authKey":"key","talkGroup":1}`, want: "callsign has the wrong type"},
+		{name: "empty auth key", body: configJSON("127.0.0.1", 5308, "DL1ABC", "", 1), want: "authKey is empty"},
+		{name: "whitespace auth key", body: configJSON("127.0.0.1", 5308, "DL1ABC", " \t", 1), want: "authKey is empty"},
+		{name: "auth key type", body: `{"host":"127.0.0.1","port":5308,"callsign":"DL1ABC","authKey":1,"talkGroup":1}`, want: "authKey has the wrong type"},
+		{name: "port zero", body: configJSON("127.0.0.1", 0, "DL1ABC", "key", 1), want: "port is invalid"},
+		{name: "port negative", body: configJSON("127.0.0.1", -1, "DL1ABC", "key", 1), want: "port is invalid"},
+		{name: "port too high", body: configJSON("127.0.0.1", 65536, "DL1ABC", "key", 1), want: "port is invalid"},
+		{name: "port type", body: `{"host":"127.0.0.1","port":"5308","callsign":"DL1ABC","authKey":"key","talkGroup":1}`, want: "port has the wrong type"},
+		{name: "port fraction", body: `{"host":"127.0.0.1","port":5308.5,"callsign":"DL1ABC","authKey":"key","talkGroup":1}`, want: "port is invalid"},
+		{name: "talk group zero", body: configJSON("127.0.0.1", 5308, "DL1ABC", "key", 0), want: "talkGroup is invalid"},
+		{name: "talk group negative", body: configJSON("127.0.0.1", 5308, "DL1ABC", "key", -3), want: "talkGroup is invalid"},
+		{name: "talk group type", body: `{"host":"127.0.0.1","port":5308,"callsign":"DL1ABC","authKey":"key","talkGroup":"20"}`, want: "talkGroup has the wrong type"},
+		{name: "bad callsign", body: configJSON("127.0.0.1", 5308, "HELLO", "key", 1), want: "callsign is invalid"},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			path := filepath.Join(t.TempDir(), "funknetz.json")
+			if err := os.WriteFile(path, []byte(tt.body), 0o600); err != nil {
+				t.Fatal(err)
+			}
+			before, err := os.ReadFile(path)
+			if err != nil {
+				t.Fatal(err)
+			}
+			_, err = station.New(station.Options{ConfigPath: path})
+			if err == nil {
+				t.Fatal("New() error = nil, want a field rejection")
+			}
+			if err.Error() != tt.want {
+				t.Fatalf("New() error = %q, want %q", err.Error(), tt.want)
+			}
+			after, err := os.ReadFile(path)
+			if err != nil {
+				t.Fatal(err)
+			}
+			if string(after) != string(before) {
+				t.Fatalf("file = %q, want it left untouched", after)
+			}
+		})
+	}
+}
+
+func TestBrokenConfigSaysTheFileCouldNotBeRead(t *testing.T) {
+	path := filepath.Join(t.TempDir(), "funknetz.json")
+	if err := os.WriteFile(path, []byte("{"), 0o600); err != nil {
+		t.Fatal(err)
+	}
+	_, err := station.New(station.Options{ConfigPath: path})
+	if err == nil {
+		t.Fatal("New() error = nil, want a read failure")
+	}
+	if err.Error() != "could not be read" {
+		t.Fatalf("New() error = %q, want %q", err.Error(), "could not be read")
+	}
+	if strings.Contains(err.Error(), "callsign") {
+		t.Fatalf("New() error = %q, want no field name", err.Error())
+	}
+}
+
+func TestCallsignIsTrimmedAndUppercased(t *testing.T) {
+	path := filepath.Join(t.TempDir(), "funknetz.json")
+	body := configJSON("  127.0.0.1  ", 5308, "  dl1abc-r  ", "a b", 20)
+	if err := os.WriteFile(path, []byte(body), 0o600); err != nil {
+		t.Fatal(err)
+	}
+	st, err := station.New(station.Options{ConfigPath: path})
+	if err != nil {
+		t.Fatal(err)
+	}
+	got := st.State()
+	if got.Callsign != "DL1ABC-R" {
+		t.Fatalf("Callsign = %q, want DL1ABC-R", got.Callsign)
+	}
+	if got.TalkGroup != 20 {
+		t.Fatalf("TalkGroup = %d, want 20", got.TalkGroup)
+	}
+	if got.Connection != "Disconnected" || got.Detail != "Idle" || got.Talker != "" || got.Transmitting {
+		t.Fatalf("State = %+v, want Disconnected, Idle, no talker", got)
+	}
+}
+
+func configJSON(host string, port int, callsign, authKey string, talkGroup int) string {
+	body, err := jsonv2.Marshal(map[string]any{
+		"host":      host,
+		"port":      port,
+		"callsign":  callsign,
+		"authKey":   authKey,
+		"talkGroup": talkGroup,
+	})
+	if err != nil {
+		panic(err)
+	}
+	return string(body)
+}

station/protocol.go

@@ -0,0 +1,158 @@
+package station
+
+import (
+	"encoding/binary"
+	"fmt"
+	"io"
+)
+
+const (
+	tcpHeartbeat     = 1
+	tcpProtoVer      = 5
+	tcpAuthChallenge = 10
+	tcpAuthResponse  = 11
+	tcpAuthOK        = 12
+	tcpError         = 13
+	tcpServerInfo    = 100
+	tcpTalkerStart   = 104
+	tcpTalkerStop    = 105
+	tcpSelectTG      = 106
+
+	udpHeartbeat = 1
+	udpAudio     = 101
+	udpFlush     = 102
+
+	maxFrame = 32768
+)
+
+type frame struct {
+	typ  uint16
+	body []byte
+}
+
+func writeFrame(w io.Writer, typ uint16, body []byte) error {
+	payload := make([]byte, 2+len(body))
+	binary.BigEndian.PutUint16(payload[:2], typ)
+	copy(payload[2:], body)
+	var hdr [4]byte
+	binary.BigEndian.PutUint32(hdr[:], uint32(len(payload)))
+	if _, err := w.Write(hdr[:]); err != nil {
+		return err
+	}
+	_, err := w.Write(payload)
+	return err
+}
+
+func readFrame(r io.Reader) (frame, error) {
+	var hdr [4]byte
+	if _, err := io.ReadFull(r, hdr[:]); err != nil {
+		return frame{}, err
+	}
+	n := binary.BigEndian.Uint32(hdr[:])
+	if n < 2 || n > maxFrame {
+		return frame{}, fmt.Errorf("frame length %d", n)
+	}
+	buf := make([]byte, n)
+	if _, err := io.ReadFull(r, buf); err != nil {
+		return frame{}, err
+	}
+	return frame{typ: binary.BigEndian.Uint16(buf[:2]), body: buf[2:]}, nil
+}
+
+func appendU16(b []byte, v uint16) []byte {
+	var buf [2]byte
+	binary.BigEndian.PutUint16(buf[:], v)
+	return append(b, buf[:]...)
+}
+
+func appendString(b []byte, s string) []byte {
+	b = appendU16(b, uint16(len(s)))
+	return append(b, s...)
+}
+
+func appendBlob(b []byte, p []byte) []byte {
+	b = appendU16(b, uint16(len(p)))
+	return append(b, p...)
+}
+
+func parseU16(b []byte) (uint16, []byte, error) {
+	if len(b) < 2 {
+		return 0, nil, io.ErrUnexpectedEOF
+	}
+	return binary.BigEndian.Uint16(b[:2]), b[2:], nil
+}
+
+func parseU32(b []byte) (uint32, []byte, error) {
+	if len(b) < 4 {
+		return 0, nil, io.ErrUnexpectedEOF
+	}
+	return binary.BigEndian.Uint32(b[:4]), b[4:], nil
+}
+
+func parseString(b []byte) (string, []byte, error) {
+	n, rest, err := parseU16(b)
+	if err != nil {
+		return "", nil, err
+	}
+	if len(rest) < int(n) {
+		return "", nil, io.ErrUnexpectedEOF
+	}
+	return string(rest[:n]), rest[n:], nil
+}
+
+func parseBlob(b []byte) ([]byte, []byte, error) {
+	n, rest, err := parseU16(b)
+	if err != nil {
+		return nil, nil, err
+	}
+	if len(rest) < int(n) {
+		return nil, nil, io.ErrUnexpectedEOF
+	}
+	return rest[:n], rest[n:], nil
+}
+
+func parseStringList(b []byte) ([]string, []byte, error) {
+	n, rest, err := parseU16(b)
+	if err != nil {
+		return nil, nil, err
+	}
+	out := make([]string, 0, n)
+	for range int(n) {
+		s, next, err := parseString(rest)
+		if err != nil {
+			return nil, nil, err
+		}
+		out = append(out, s)
+		rest = next
+	}
+	return out, rest, nil
+}
+
+func parseServerInfo(b []byte) (uint16, []string, error) {
+	_, rest, err := parseU16(b)
+	if err != nil {
+		return 0, nil, err
+	}
+	id, rest, err := parseU16(rest)
+	if err != nil {
+		return 0, nil, err
+	}
+	_, rest, err = parseStringList(rest)
+	if err != nil {
+		return 0, nil, err
+	}
+	codecs, _, err := parseStringList(rest)
+	if err != nil {
+		return 0, nil, err
+	}
+	return id, codecs, nil
+}
+
+func hasOPUS(codecs []string) bool {
+	for _, c := range codecs {
+		if c == "OPUS" {
+			return true
+		}
+	}
+	return false
+}

station/reflector_test.go

@@ -0,0 +1,325 @@
+package station_test
+
+import (
+	"encoding/binary"
+	"io"
+	"net"
+	"os"
+	"sync"
+	"testing"
+	"time"
+
+	"gitea.neitzel.de/konrad/funknetz/station"
+)
+
+type fakeClock struct {
+	mu      sync.Mutex
+	now     time.Time
+	waiters []*fakeWaiter
+	gen     int
+}
+
+type fakeWaiter struct {
+	deadline time.Time
+	ch       chan time.Time
+	done     bool
+}
+
+func newFakeClock() *fakeClock {
+	return &fakeClock{now: time.Date(2026, 10, 3, 12, 0, 0, 0, time.UTC)}
+}
+
+func (c *fakeClock) Now() time.Time {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	return c.now
+}
+
+func (c *fakeClock) NewTimer(d time.Duration) station.Timer {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	if d < 0 {
+		d = 0
+	}
+	ch := make(chan time.Time, 1)
+	c.gen++
+	w := &fakeWaiter{deadline: c.now.Add(d), ch: ch}
+	if d == 0 {
+		ch <- c.now
+		w.done = true
+		return &fakeTimer{c: c, w: w}
+	}
+	c.waiters = append(c.waiters, w)
+	return &fakeTimer{c: c, w: w}
+}
+
+func (c *fakeClock) Advance(d time.Duration) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	c.now = c.now.Add(d)
+	kept := c.waiters[:0]
+	for _, w := range c.waiters {
+		if w.done {
+			continue
+		}
+		if !w.deadline.After(c.now) {
+			w.done = true
+			w.ch <- c.now
+			continue
+		}
+		kept = append(kept, w)
+	}
+	c.waiters = kept
+}
+
+func (c *fakeClock) Generation() int {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	return c.gen
+}
+
+func waitClock(t *testing.T, c *fakeClock, after int) {
+	t.Helper()
+	deadline := time.Now().Add(2 * time.Second)
+	for c.Generation() <= after {
+		if time.Now().After(deadline) {
+			t.Fatal("station did not rearm the clock")
+		}
+		time.Sleep(time.Millisecond)
+	}
+}
+
+func (c *fakeClock) BlockUntilWaiting(t *testing.T) {
+	t.Helper()
+	deadline := time.Now().Add(2 * time.Second)
+	for {
+		c.mu.Lock()
+		n := 0
+		for _, w := range c.waiters {
+			if !w.done {
+				n++
+			}
+		}
+		c.mu.Unlock()
+		if n > 0 {
+			return
+		}
+		if time.Now().After(deadline) {
+			t.Fatal("station did not wait on the clock")
+		}
+		time.Sleep(time.Millisecond)
+	}
+}
+
+type fakeTimer struct {
+	c *fakeClock
+	w *fakeWaiter
+}
+
+func (t *fakeTimer) C() <-chan time.Time { return t.w.ch }
+
+func (t *fakeTimer) Stop() bool {
+	t.c.mu.Lock()
+	defer t.c.mu.Unlock()
+	if t.w.done {
+		return false
+	}
+	t.w.done = true
+	return true
+}
+
+type reflector struct {
+	ln   net.Listener
+	udp  *net.UDPConn
+	tcp  net.Conn
+	peer *net.UDPAddr
+	port int
+}
+
+func newReflector(t *testing.T) *reflector {
+	t.Helper()
+	ln, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatal(err)
+	}
+	port := ln.Addr().(*net.TCPAddr).Port
+	udp, err := net.ListenUDP("udp", &net.UDPAddr{IP: net.ParseIP("127.0.0.1"), Port: port})
+	if err != nil {
+		ln.Close()
+		t.Fatal(err)
+	}
+	t.Cleanup(func() {
+		if ln != nil {
+			ln.Close()
+		}
+		udp.Close()
+	})
+	return &reflector{ln: ln, udp: udp, port: port}
+}
+
+func (r *reflector) accept(t *testing.T) {
+	t.Helper()
+	r.ln.(*net.TCPListener).SetDeadline(time.Now().Add(2 * time.Second))
+	conn, err := r.ln.Accept()
+	if err != nil {
+		t.Fatal(err)
+	}
+	r.tcp = conn
+	t.Cleanup(func() { conn.Close() })
+}
+
+func (r *reflector) readFrame(t *testing.T) (uint16, []byte) {
+	t.Helper()
+	if err := r.tcp.SetReadDeadline(time.Now().Add(2 * time.Second)); err != nil {
+		t.Fatal(err)
+	}
+	var hdr [4]byte
+	if _, err := io.ReadFull(r.tcp, hdr[:]); err != nil {
+		t.Fatal(err)
+	}
+	n := binary.BigEndian.Uint32(hdr[:])
+	if n > 32768 {
+		t.Fatalf("frame length %d", n)
+	}
+	buf := make([]byte, n)
+	if _, err := io.ReadFull(r.tcp, buf); err != nil {
+		t.Fatal(err)
+	}
+	if len(buf) < 2 {
+		t.Fatalf("short frame %d", len(buf))
+	}
+	return binary.BigEndian.Uint16(buf[:2]), buf[2:]
+}
+
+func (r *reflector) writeFrame(t *testing.T, typ uint16, body []byte) {
+	t.Helper()
+	payload := make([]byte, 2+len(body))
+	binary.BigEndian.PutUint16(payload, typ)
+	copy(payload[2:], body)
+	var hdr [4]byte
+	binary.BigEndian.PutUint32(hdr[:], uint32(len(payload)))
+	if err := r.tcp.SetWriteDeadline(time.Now().Add(2 * time.Second)); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := r.tcp.Write(append(hdr[:], payload...)); err != nil {
+		t.Fatal(err)
+	}
+}
+
+func (r *reflector) readUDP(t *testing.T) (typ, id, seq uint16, body []byte) {
+	t.Helper()
+	if err := r.udp.SetReadDeadline(time.Now().Add(2 * time.Second)); err != nil {
+		t.Fatal(err)
+	}
+	buf := make([]byte, 2048)
+	n, addr, err := r.udp.ReadFromUDP(buf)
+	if err != nil {
+		t.Fatal(err)
+	}
+	r.peer = addr
+	if n < 6 {
+		t.Fatalf("short datagram %d", n)
+	}
+	typ = binary.BigEndian.Uint16(buf[0:2])
+	id = binary.BigEndian.Uint16(buf[2:4])
+	seq = binary.BigEndian.Uint16(buf[4:6])
+	body = append([]byte(nil), buf[6:n]...)
+	return typ, id, seq, body
+}
+
+func (r *reflector) writeUDP(t *testing.T, typ, id, seq uint16, body []byte) {
+	t.Helper()
+	if r.peer == nil {
+		t.Fatal("no udp peer")
+	}
+	buf := make([]byte, 6+len(body))
+	binary.BigEndian.PutUint16(buf[0:2], typ)
+	binary.BigEndian.PutUint16(buf[2:4], id)
+	binary.BigEndian.PutUint16(buf[4:6], seq)
+	copy(buf[6:], body)
+	if err := r.udp.SetWriteDeadline(time.Now().Add(2 * time.Second)); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := r.udp.WriteToUDP(buf, r.peer); err != nil {
+		t.Fatal(err)
+	}
+}
+
+func appendU16(b []byte, v uint16) []byte {
+	var buf [2]byte
+	binary.BigEndian.PutUint16(buf[:], v)
+	return append(b, buf[:]...)
+}
+
+func appendU32(b []byte, v uint32) []byte {
+	var buf [4]byte
+	binary.BigEndian.PutUint32(buf[:], v)
+	return append(b, buf[:]...)
+}
+
+func appendStr(b []byte, s string) []byte {
+	b = appendU16(b, uint16(len(s)))
+	return append(b, s...)
+}
+
+func appendBytes(b []byte, p []byte) []byte {
+	b = appendU16(b, uint16(len(p)))
+	return append(b, p...)
+}
+
+func takeU16(t *testing.T, b []byte) (uint16, []byte) {
+	t.Helper()
+	if len(b) < 2 {
+		t.Fatalf("need 2 bytes, have %d", len(b))
+	}
+	return binary.BigEndian.Uint16(b[:2]), b[2:]
+}
+
+func takeU32(t *testing.T, b []byte) (uint32, []byte) {
+	t.Helper()
+	if len(b) < 4 {
+		t.Fatalf("need 4 bytes, have %d", len(b))
+	}
+	return binary.BigEndian.Uint32(b[:4]), b[4:]
+}
+
+func takeStr(t *testing.T, b []byte) (string, []byte) {
+	t.Helper()
+	n, rest := takeU16(t, b)
+	if len(rest) < int(n) {
+		t.Fatalf("string needs %d bytes, have %d", n, len(rest))
+	}
+	return string(rest[:n]), rest[n:]
+}
+
+func takeBytes(t *testing.T, b []byte) ([]byte, []byte) {
+	t.Helper()
+	n, rest := takeU16(t, b)
+	if len(rest) < int(n) {
+		t.Fatalf("bytes need %d, have %d", n, len(rest))
+	}
+	return append([]byte(nil), rest[:n]...), rest[n:]
+}
+
+func serverInfo(clientID uint16, nodes, codecs []string) []byte {
+	b := appendU16(nil, 0)
+	b = appendU16(b, clientID)
+	b = appendU16(b, uint16(len(nodes)))
+	for _, n := range nodes {
+		b = appendStr(b, n)
+	}
+	b = appendU16(b, uint16(len(codecs)))
+	for _, c := range codecs {
+		b = appendStr(b, c)
+	}
+	return b
+}
+
+func writeConfigFile(t *testing.T, body string) string {
+	t.Helper()
+	path := t.TempDir() + "/funknetz.json"
+	if err := os.WriteFile(path, []byte(body), 0o600); err != nil {
+		t.Fatal(err)
+	}
+	return path
+}

station/session.go

@@ -0,0 +1,522 @@
+package station
+
+import (
+	"context"
+	"crypto/hmac"
+	"crypto/sha1"
+	"encoding/binary"
+	"errors"
+	"net"
+	"strconv"
+	"sync"
+	"time"
+
+	"gitea.neitzel.de/konrad/funknetz/internal/opus"
+)
+
+const noOpus = "The reflector did not offer OPUS."
+
+var errStopped = errors.New("session stopped")
+
+// Run keeps the reflector session until ctx is canceled.
+func (s *Station) Run(ctx context.Context) error {
+	for {
+		if ctx.Err() != nil {
+			return nil
+		}
+		s.mu.Lock()
+		if s.fatal != "" {
+			s.mu.Unlock()
+			<-ctx.Done()
+			return nil
+		}
+		s.state.Connection = "Connecting"
+		s.publish()
+		host, port := s.cfg.host, s.cfg.port
+		s.mu.Unlock()
+
+		conn, err := dialStation(ctx, host, port)
+		if err != nil {
+			if ctx.Err() != nil {
+				return nil
+			}
+			s.mu.Lock()
+			s.state.Connection = "Disconnected"
+			s.publish()
+			s.mu.Unlock()
+			if !s.wait(ctx, 2*time.Second) {
+				return nil
+			}
+			continue
+		}
+
+		err = s.session(ctx, conn)
+		conn.Close()
+		s.mu.Lock()
+		s.closeUDP()
+		if s.fatal == "" {
+			s.state.Connection = "Disconnected"
+			s.remoteTalker = ""
+			s.roomBusy = false
+		}
+		s.mic = nil
+		s.play = nil
+		s.publish()
+		fatal := s.fatal != ""
+		s.mu.Unlock()
+		if ctx.Err() != nil {
+			return nil
+		}
+		if fatal {
+			<-ctx.Done()
+			return nil
+		}
+		if err != nil && errors.Is(err, errStopped) {
+			<-ctx.Done()
+			return nil
+		}
+		if !s.wait(ctx, 2*time.Second) {
+			return nil
+		}
+	}
+}
+
+func dialStation(ctx context.Context, host string, port int) (net.Conn, error) {
+	var d net.Dialer
+	d.Timeout = 5 * time.Second
+	return d.DialContext(ctx, "tcp", net.JoinHostPort(host, strconv.Itoa(port)))
+}
+
+func (s *Station) wait(ctx context.Context, d time.Duration) bool {
+	timer := s.clock.NewTimer(d)
+	defer timer.Stop()
+	select {
+	case <-ctx.Done():
+		return false
+	case <-timer.C():
+		return true
+	}
+}
+
+func (s *Station) session(ctx context.Context, conn net.Conn) error {
+	done := make(chan struct{})
+	var closeOnce sync.Once
+	closeConn := func() { closeOnce.Do(func() { conn.Close() }) }
+	defer close(done)
+	go func() {
+		select {
+		case <-ctx.Done():
+			closeConn()
+		case <-done:
+		}
+	}()
+
+	tcpCh := make(chan frame, 8)
+	tcpErr := make(chan error, 1)
+	go readTCP(conn, tcpCh, tcpErr, done)
+
+	if err := writeFrame(conn, tcpProtoVer, []byte{0, 2, 0, 0}); err != nil {
+		return err
+	}
+	lastTCP := s.clock.Now()
+	tcpRx := lastTCP.Add(15 * time.Second)
+
+	var udpCh chan []byte
+	var udpErr chan error
+	udpReady := false
+	var udpRx time.Time
+
+	for {
+		s.mu.Lock()
+		lastUDP := s.lastUDPSend
+		s.mu.Unlock()
+
+		next := lastTCP.Add(10 * time.Second)
+		if tcpRx.Before(next) {
+			next = tcpRx
+		}
+		if udpReady {
+			if u := lastUDP.Add(15 * time.Second); u.Before(next) {
+				next = u
+			}
+			if udpRx.Before(next) {
+				next = udpRx
+			}
+		}
+		wait := next.Sub(s.clock.Now())
+		if wait < 0 {
+			wait = 0
+		}
+		timer := s.clock.NewTimer(wait)
+
+		var udpRecv <-chan []byte
+		var udpFail <-chan error
+		if udpCh != nil {
+			udpRecv = udpCh
+			udpFail = udpErr
+		}
+
+		var reason string
+		var tcpf frame
+		var udpp []byte
+		select {
+		case <-ctx.Done():
+			timer.Stop()
+			return ctx.Err()
+		case <-tcpErr:
+			timer.Stop()
+			return errors.New("tcp closed")
+		case <-udpFail:
+			timer.Stop()
+			return errors.New("udp closed")
+		case tcpf = <-tcpCh:
+			timer.Stop()
+			reason = "tcp"
+		case udpp = <-udpRecv:
+			timer.Stop()
+			reason = "udp"
+		case <-timer.C():
+			select {
+			case tcpf = <-tcpCh:
+				reason = "tcp"
+			default:
+				select {
+				case udpp = <-udpRecv:
+					reason = "udp"
+				default:
+					reason = "timer"
+				}
+			}
+		}
+
+		now := s.clock.Now()
+		switch reason {
+		case "tcp":
+			tcpRx = now.Add(15 * time.Second)
+			if err := s.onTCP(conn, tcpf, done, &udpCh, &udpErr, &udpReady, &udpRx, &lastTCP); err != nil {
+				return err
+			}
+		case "udp":
+			s.mu.Lock()
+			if clientIDOf(udpp) == s.clientID {
+				udpRx = now.Add(60 * time.Second)
+				s.onDatagram(udpp)
+			}
+			s.mu.Unlock()
+		default:
+			if !now.Before(tcpRx) {
+				return errors.New("tcp receive timeout")
+			}
+			if udpReady && !now.Before(udpRx) {
+				return errors.New("udp receive timeout")
+			}
+			if !now.Before(lastTCP.Add(10 * time.Second)) {
+				if err := writeFrame(conn, tcpHeartbeat, nil); err != nil {
+					return err
+				}
+				lastTCP = s.clock.Now()
+			}
+			s.mu.Lock()
+			if udpReady && s.udp != nil && !s.clock.Now().Before(s.lastUDPSend.Add(15*time.Second)) {
+				if err := s.sendUDP(udpHeartbeat, nil); err != nil {
+					s.mu.Unlock()
+					return err
+				}
+			}
+			s.mu.Unlock()
+		}
+	}
+}
+
+func readTCP(conn net.Conn, out chan<- frame, errc chan<- error, done <-chan struct{}) {
+	for {
+		f, err := readFrame(conn)
+		if err != nil {
+			select {
+			case errc <- err:
+			case <-done:
+			}
+			return
+		}
+		select {
+		case out <- f:
+		case <-done:
+			return
+		}
+	}
+}
+
+func readUDP(conn *net.UDPConn, out chan<- []byte, errc chan<- error, done <-chan struct{}) {
+	buf := make([]byte, 2048)
+	for {
+		n, err := conn.Read(buf)
+		if err != nil {
+			select {
+			case errc <- err:
+			case <-done:
+			}
+			return
+		}
+		pkt := append([]byte(nil), buf[:n]...)
+		select {
+		case out <- pkt:
+		case <-done:
+			return
+		}
+	}
+}
+
+func (s *Station) onTCP(conn net.Conn, f frame, done <-chan struct{}, udpCh *chan []byte, udpErr *chan error, udpReady *bool, udpRx *time.Time, lastTCP *time.Time) error {
+	switch f.typ {
+	case tcpHeartbeat, tcpAuthOK:
+		return nil
+	case tcpAuthChallenge:
+		chal, _, err := parseBlob(f.body)
+		if err != nil || len(chal) != 20 {
+			return errors.New("bad challenge")
+		}
+		mac := hmac.New(sha1.New, []byte(s.cfg.authKey))
+		mac.Write(chal)
+		body := appendString(nil, s.cfg.callsign)
+		body = appendBlob(body, mac.Sum(nil))
+		if err := writeFrame(conn, tcpAuthResponse, body); err != nil {
+			return err
+		}
+		*lastTCP = s.clock.Now()
+		return nil
+	case tcpError:
+		msg, _, err := parseString(f.body)
+		if err != nil {
+			msg = "could not be read"
+		}
+		s.stop(msg)
+		return errStopped
+	case tcpServerInfo:
+		return s.onServerInfo(conn, f.body, done, udpCh, udpErr, udpReady, udpRx, lastTCP)
+	case tcpTalkerStart:
+		s.onTalker(f.body, true)
+		return nil
+	case tcpTalkerStop:
+		s.onTalker(f.body, false)
+		return nil
+	default:
+		return nil
+	}
+}
+
+func (s *Station) onServerInfo(conn net.Conn, body []byte, done <-chan struct{}, udpCh *chan []byte, udpErr *chan error, udpReady *bool, udpRx *time.Time, lastTCP *time.Time) error {
+	id, codecs, err := parseServerInfo(body)
+	if err != nil {
+		return err
+	}
+	if !hasOPUS(codecs) {
+		s.stop(noOpus)
+		return errStopped
+	}
+	enc, err := opus.NewEncoder()
+	if err != nil {
+		return err
+	}
+	dec, err := opus.NewDecoder()
+	if err != nil {
+		enc.Close()
+		return err
+	}
+	var tg [4]byte
+	binary.BigEndian.PutUint32(tg[:], uint32(s.cfg.talkGroup))
+	if err := writeFrame(conn, tcpSelectTG, tg[:]); err != nil {
+		enc.Close()
+		dec.Close()
+		return err
+	}
+	*lastTCP = s.clock.Now()
+
+	remote, ok := conn.RemoteAddr().(*net.TCPAddr)
+	if !ok {
+		enc.Close()
+		dec.Close()
+		return errors.New("tcp peer has no address")
+	}
+	udpConn, err := net.DialUDP("udp", nil, &net.UDPAddr{IP: remote.IP, Port: remote.Port})
+	if err != nil {
+		enc.Close()
+		dec.Close()
+		return err
+	}
+	ch := make(chan []byte, 32)
+	errc := make(chan error, 1)
+	go readUDP(udpConn, ch, errc, done)
+
+	s.mu.Lock()
+	s.clientID = id
+	s.seq = 0
+	s.udp = udpConn
+	s.enc = enc
+	s.dec = dec
+	err = s.sendUDP(udpHeartbeat, nil)
+	if err != nil {
+		s.mu.Unlock()
+		enc.Close()
+		dec.Close()
+		udpConn.Close()
+		return err
+	}
+	s.state.Connection = "Connected"
+	s.publish()
+	s.mu.Unlock()
+
+	*udpCh = ch
+	*udpErr = errc
+	*udpReady = true
+	*udpRx = s.clock.Now().Add(60 * time.Second)
+	return nil
+}
+
+func (s *Station) stop(msg string) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.fatal = msg
+	s.publish()
+}
+
+// Fail stops retries and shows message until the operator quits.
+func (s *Station) Fail(message string) { s.stop(message) }
+
+func (s *Station) onTalker(body []byte, start bool) {
+	tg, rest, err := parseU32(body)
+	if err != nil || int(tg) != s.cfg.talkGroup {
+		return
+	}
+	call, _, err := parseString(rest)
+	if err != nil {
+		return
+	}
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	if start {
+		if call == s.cfg.callsign {
+			return
+		}
+		was := s.transmitting()
+		s.remoteTalker = call
+		s.roomBusy = true
+		if was && !s.transmitting() {
+			s.endTransmit()
+		}
+		s.publish()
+		return
+	}
+	s.remoteTalker = ""
+	s.roomBusy = false
+	s.publish()
+}
+
+func (s *Station) onDatagram(pkt []byte) {
+	if len(pkt) < 6 {
+		return
+	}
+	typ := binary.BigEndian.Uint16(pkt[:2])
+	if typ != udpAudio || s.dec == nil {
+		return
+	}
+	if s.transmitting() {
+		return
+	}
+	packet, _, err := parseBlob(pkt[6:])
+	if err != nil {
+		return
+	}
+	pcm, err := s.dec.Decode(packet)
+	if err != nil {
+		return
+	}
+	s.play = append(s.play, pcm...)
+	s.roomBusy = true
+	s.publish()
+}
+
+func clientIDOf(pkt []byte) uint16 {
+	if len(pkt) < 6 {
+		return ^uint16(0)
+	}
+	return binary.BigEndian.Uint16(pkt[2:4])
+}
+
+func (s *Station) sendUDP(typ uint16, body []byte) error {
+	if s.udp == nil {
+		return errors.New("udp is down")
+	}
+	buf := make([]byte, 6+len(body))
+	binary.BigEndian.PutUint16(buf[0:2], typ)
+	binary.BigEndian.PutUint16(buf[2:4], s.clientID)
+	binary.BigEndian.PutUint16(buf[4:6], s.seq)
+	s.seq++
+	copy(buf[6:], body)
+	if _, err := s.udp.Write(buf); err != nil {
+		return err
+	}
+	s.lastUDPSend = s.clock.Now()
+	return nil
+}
+
+func (s *Station) endTransmit() {
+	if s.enc != nil && s.udp != nil && len(s.mic) > 0 {
+		frame := make([]int16, opus.FrameSamples)
+		copy(frame, s.mic)
+		s.mic = nil
+		if pkt, err := s.enc.Encode(frame); err == nil {
+			_ = s.sendUDP(udpAudio, appendBlob(nil, pkt))
+		}
+	} else {
+		s.mic = nil
+	}
+	if s.udp != nil {
+		_ = s.sendUDP(udpFlush, nil)
+	}
+}
+
+func (s *Station) closeUDP() {
+	if s.udp != nil {
+		s.udp.Close()
+		s.udp = nil
+	}
+	if s.enc != nil {
+		s.enc.Close()
+		s.enc = nil
+	}
+	if s.dec != nil {
+		s.dec.Close()
+		s.dec = nil
+	}
+}
+
+func (s *Station) transmitting() bool {
+	if s.fatal != "" || s.state.Connection != "Connected" || s.roomBusy {
+		return false
+	}
+	return s.buttonHeld || s.spaceHeld
+}
+
+func (s *Station) publish() {
+	s.state.Callsign = s.cfg.callsign
+	s.state.TalkGroup = s.cfg.talkGroup
+	if s.fatal != "" {
+		s.state.Connection = "Disconnected"
+		s.state.Detail = s.fatal
+		s.state.Talker = ""
+		s.state.Transmitting = false
+	} else {
+		s.state.Detail = "Idle"
+		tx := s.transmitting()
+		s.state.Transmitting = tx
+		if tx {
+			s.state.Talker = s.cfg.callsign
+		} else {
+			s.state.Talker = s.remoteTalker
+		}
+	}
+	select {
+	case s.wake <- struct{}{}:
+	default:
+	}
+}

station/session_test.go

@@ -0,0 +1,99 @@
+package station_test
+
+import (
+	"bytes"
+	"context"
+	"crypto/hmac"
+	"crypto/sha1"
+	"testing"
+	"time"
+
+	"gitea.neitzel.de/konrad/funknetz/station"
+)
+
+func TestLoginAnnouncesProtocolAndJoinsTheTalkGroup(t *testing.T) {
+	refl := newReflector(t)
+	path := writeConfigFile(t, configJSON("  127.0.0.1  ", refl.port, "  dl1abc  ", "auth key with spaces", 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) }()
+
+	got, err := st.Wait(ctx, func(s station.State) bool { return s.Connection == "Connecting" })
+	if err != nil {
+		t.Fatal(err)
+	}
+	if got.Callsign != "DL1ABC" || got.TalkGroup != 20 || got.Detail != "Idle" {
+		t.Fatalf("State = %+v, want DL1ABC talk group 20 Idle", got)
+	}
+
+	refl.accept(t)
+	typ, body := refl.readFrame(t)
+	if typ != 5 || len(body) != 4 || body[0] != 0 || body[1] != 2 || body[2] != 0 || body[3] != 0 {
+		t.Fatalf("first frame type %d body %v, want protocol 2.0", 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 response type %d, want 11", typ)
+	}
+	call, rest := takeStr(t, body)
+	digest, rest := takeBytes(t, rest)
+	if len(rest) != 0 {
+		t.Fatalf("auth response has %d trailing bytes", len(rest))
+	}
+	if call != "DL1ABC" {
+		t.Fatalf("callsign = %q, want DL1ABC", call)
+	}
+	mac := hmac.New(sha1.New, []byte("auth key with spaces"))
+	mac.Write(challenge)
+	if !hmac.Equal(digest, mac.Sum(nil)) {
+		t.Fatalf("digest = %x, want hmac-sha1 of the challenge under the stored key", digest)
+	}
+	if bytes.Contains(body, []byte("auth key with spaces")) {
+		t.Fatal("auth response contains the auth key")
+	}
+
+	refl.writeFrame(t, 12, nil)
+	refl.writeFrame(t, 100, serverInfo(0x1234, []string{"db0abc"}, []string{"GSM", "OPUS"}))
+
+	typ, body = refl.readFrame(t)
+	if typ != 106 {
+		t.Fatalf("frame type %d, want select talk group", typ)
+	}
+	tg, rest := takeU32(t, body)
+	if tg != 20 || len(rest) != 0 {
+		t.Fatalf("select talk group = %d rest %d, want 20", tg, len(rest))
+	}
+
+	utyp, id, seq, ubody := refl.readUDP(t)
+	if utyp != 1 || id != 0x1234 || seq != 0 || len(ubody) != 0 {
+		t.Fatalf("udp heartbeat type %d id %d seq %d len %d, want type 1 id 0x1234 seq 0", utyp, id, seq, len(ubody))
+	}
+
+	got, err = st.Wait(ctx, func(s station.State) bool { return s.Connection == "Connected" })
+	if err != nil {
+		t.Fatal(err)
+	}
+	if got.Detail != "Idle" || got.Talker != "" {
+		t.Fatalf("State = %+v, want Connected Idle", got)
+	}
+
+	// No other talk group and no station profile: the next TCP message is a heartbeat, not type 107 or 111.
+	clock.BlockUntilWaiting(t)
+	clock.Advance(10 * time.Second)
+	typ, _ = refl.readFrame(t)
+	if typ != 1 {
+		t.Fatalf("quiet TCP frame type %d, want heartbeat 1", typ)
+	}
+}

station/station.go

@@ -0,0 +1,184 @@
+// Package station is the headset station. It loads the config, keeps one
+// reflector session, and exposes the state the window shows.
+package station
+
+import (
+	"context"
+	"net"
+	"sync"
+	"time"
+
+	"gitea.neitzel.de/konrad/funknetz/internal/opus"
+)
+
+// Options selects the config file and the clock used for heartbeats and retries.
+type Options struct {
+	ConfigPath string
+	Clock      Clock
+}
+
+// State is what the window shows.
+type State struct {
+	Connection   string `json:"connection"`
+	Detail       string `json:"detail"`
+	Callsign     string `json:"callsign"`
+	TalkGroup    int    `json:"talkGroup"`
+	Talker       string `json:"talker"`
+	Transmitting bool   `json:"transmitting"`
+}
+
+// Station is the seam tests and the window use.
+type Station struct {
+	cfg   loaded
+	clock Clock
+
+	mu           sync.Mutex
+	state        State
+	wake         chan struct{}
+	buttonHeld   bool
+	spaceHeld    bool
+	fatal        string
+	remoteTalker string
+	roomBusy     bool
+	mic          []int16
+	play         []int16
+	clientID     uint16
+	seq          uint16
+	udp          *net.UDPConn
+	lastUDPSend  time.Time
+	enc          *opus.Encoder
+	dec          *opus.Decoder
+}
+
+// New loads the config at opt.ConfigPath. A missing file is created mode 0600
+// and New returns an error so the caller can exit before opening a window.
+func New(opt Options) (*Station, error) {
+	cfg, err := loadConfig(opt.ConfigPath)
+	if err != nil {
+		return nil, err
+	}
+	clock := opt.Clock
+	if clock == nil {
+		clock = wallClock{}
+	}
+	s := &Station{
+		cfg:   cfg,
+		clock: clock,
+		wake:  make(chan struct{}, 1),
+		state: State{
+			Connection: "Disconnected",
+			Detail:     "Idle",
+			Callsign:   cfg.callsign,
+			TalkGroup:  cfg.talkGroup,
+		},
+	}
+	return s, nil
+}
+
+// State returns the latest window state.
+func (s *Station) State() State {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	return s.state
+}
+
+// Wait blocks until pred is true or ctx ends.
+func (s *Station) Wait(ctx context.Context, pred func(State) bool) (State, error) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	for !pred(s.state) {
+		s.mu.Unlock()
+		select {
+		case <-ctx.Done():
+			s.mu.Lock()
+			return State{}, ctx.Err()
+		case <-s.wake:
+			s.mu.Lock()
+		}
+	}
+	return s.state, nil
+}
+
+// HoldButton reports that the on-screen button is held.
+func (s *Station) HoldButton() { s.setButton(true) }
+
+// ReleaseButton reports that the on-screen button was released.
+func (s *Station) ReleaseButton() { s.setButton(false) }
+
+// HoldSpace reports that Space is held.
+func (s *Station) HoldSpace() { s.setSpace(true) }
+
+// ReleaseSpace reports that Space was released.
+func (s *Station) ReleaseSpace() { s.setSpace(false) }
+
+func (s *Station) setButton(held bool) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	was := s.transmitting()
+	s.buttonHeld = held
+	s.afterHold(was)
+}
+
+func (s *Station) setSpace(held bool) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	was := s.transmitting()
+	s.spaceHeld = held
+	s.afterHold(was)
+}
+
+func (s *Station) afterHold(was bool) {
+	if was && !s.transmitting() {
+		s.endTransmit()
+	}
+	s.publish()
+}
+
+// WriteMicrophone accepts 16 kHz mono samples. Samples are sent only while
+// this station is allowed to talk; otherwise they are discarded.
+func (s *Station) WriteMicrophone(pcm []int16) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	if !s.transmitting() || s.enc == nil || s.udp == nil {
+		s.mic = nil
+		return
+	}
+	s.mic = append(s.mic, pcm...)
+	for len(s.mic) >= opus.FrameSamples {
+		frame := append([]int16(nil), s.mic[:opus.FrameSamples]...)
+		s.mic = append([]int16(nil), s.mic[opus.FrameSamples:]...)
+		pkt, err := s.enc.Encode(frame)
+		if err != nil {
+			continue
+		}
+		_ = s.sendUDP(udpAudio, appendBlob(nil, pkt))
+	}
+}
+
+// TakePlayback returns PCM queued for the speaker since the last take.
+func (s *Station) TakePlayback() []int16 {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	out := s.play
+	s.play = nil
+	return out
+}
+
+// WaitPlayback waits until at least n samples are queued, then returns those samples.
+func (s *Station) WaitPlayback(ctx context.Context, n int) ([]int16, error) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	for len(s.play) < n {
+		s.mu.Unlock()
+		select {
+		case <-ctx.Done():
+			s.mu.Lock()
+			return nil, ctx.Err()
+		case <-s.wake:
+			s.mu.Lock()
+		}
+	}
+	out := append([]int16(nil), s.play[:n]...)
+	s.play = append([]int16(nil), s.play[n:]...)
+	return out, nil
+}