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 path string } 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(t, "127.0.0.1", refl.port, callsign, key, tg) if logPath != "" { file = configWithLog(t, "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() { // Run's error is observed through State, which is the station seam. _ = st.Run(ctx) }() refl.accept(t) expectVersion(t, refl, 2, 0) 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)) } got, err := st.Wait(ctx, func(s station.State) bool { return s.Connection == "Connected" }) if err != nil { t.Fatal(err) } if got.Protocol != "2.0" { t.Fatalf("Protocol = %q, want 2.0", got.Protocol) } return loggedIn{st: st, clock: clock, refl: refl, cancel: cancel, id: id, path: path} } func expectVersion(t *testing.T, refl *reflector, major, minor uint16) { t.Helper() typ, body := refl.readFrame(t) if typ != 5 || !versionIs(body, major, minor) { t.Fatalf("version type %d body %v, want %d.%d", typ, body, major, minor) } } func declineTo2(t *testing.T, refl *reflector) { t.Helper() typ, body := refl.readFrame(t) if typ != 5 || !versionIs(body, 3, 0) { t.Fatalf("first version type %d body %v, want 3.0", typ, body) } refl.writeFrame(t, 6, versionWire(2, 0)) refl.accept(t) typ, body = refl.readFrame(t) if typ != 5 || !versionIs(body, 2, 0) { t.Fatalf("fallback version type %d body %v, want 2.0", typ, body) } } func versionWire(major, minor uint16) []byte { return appendU16(appendU16(nil, major), minor) } func versionIs(body []byte, major, minor uint16) bool { if len(body) != 4 { return false } return binary.BigEndian.Uint16(body[0:2]) == major && binary.BigEndian.Uint16(body[2:4]) == minor } 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) } if got := in.st.State(); got.Protocol != "" { t.Fatalf("Protocol = %q, want blank while disconnected", got.Protocol) } in.clock.BlockUntilWaiting(t) in.clock.Advance(2 * time.Second) in.refl.accept(t) typ, body := in.refl.readFrame(t) if typ != 5 || !versionIs(body, 2, 0) { t.Fatalf("retry protocol type %d body %v, want 2.0", 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(t, "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() { // Run's error is observed through State, which is the station seam. _ = 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(t, "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() { // Run's error is observed through State, which is the station seam. _ = 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(t, "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() { // Run's error is observed through State, which is the station seam. _ = st.Run(ctx) }() refl.accept(t) expectVersion(t, refl, 2, 0) 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) { conn := r.conn() _ = conn.SetReadDeadline(time.Now().Add(30 * time.Millisecond)) var hdr [4]byte if _, err := io.ReadFull(conn, hdr[:]); err != nil { return 0, nil, false } n := binary.BigEndian.Uint32(hdr[:]) buf := make([]byte, n) if _, err := io.ReadFull(conn, 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 } } }