funknetz/station/session.go

package station

import (
	"bytes"
	"context"
	"crypto/hmac"
	"crypto/rand"
	"crypto/sha1"
	"crypto/tls"
	"encoding/binary"
	"errors"
	"io"
	"net"
	"os"
	"path/filepath"
	"strconv"
	"sync"
	"sync/atomic"
	"time"

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

const noOpus = "The reflector did not offer OPUS."

var (
	errStopped   = errors.New("session stopped")
	errFallback  = errors.New("fallback")
	errReconnect = errors.New("reconnect")
)

// Run keeps the reflector session until ctx is canceled.
func (s *Station) Run(ctx context.Context) error {
	defer s.Close()
	major := uint16(3)
	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
			}
			major = 3
			s.mu.Lock()
			s.noteDownLocked(s.clock.Now(), true)
			s.mu.Unlock()
			if !s.wait(ctx, 2*time.Second) {
				return nil
			}
			continue
		}

		err = s.session(ctx, conn, major)
		conn.Close()
		fallback := errors.Is(err, errFallback)
		reconnect := errors.Is(err, errReconnect)
		s.mu.Lock()
		s.closeUDP()
		s.mic = nil
		s.play = nil
		s.audioHeard = false
		s.flushWaiting = false
		if fallback || reconnect {
			s.roomBusy = false
			s.remoteTalker = ""
			s.mu.Unlock()
			if ctx.Err() != nil {
				return nil
			}
			if fallback {
				major = 2
			} else {
				major = 3
			}
			continue
		}
		if s.fatal == "" {
			s.noteDownLocked(s.clock.Now(), false)
		} else {
			s.publish()
		}
		fatal := s.fatal != ""
		s.mu.Unlock()
		major = 3
		if ctx.Err() != nil {
			return nil
		}
		if fatal || 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
	}
}

type framed struct {
	conn    net.Conn
	mu      sync.Mutex
	w       io.Writer
	frames  chan frame
	errs    chan error
	upgrade chan *tls.Config
	up      chan error
	max     atomic.Uint32
}

func newFramed(conn net.Conn, max uint32, done <-chan struct{}) *framed {
	f := &framed{
		conn:    conn,
		w:       conn,
		frames:  make(chan frame, 8),
		errs:    make(chan error, 1),
		upgrade: make(chan *tls.Config, 1),
		up:      make(chan error, 1),
	}
	f.max.Store(max)
	go f.read(done)
	return f
}

func (f *framed) Write(p []byte) (int, error) {
	f.mu.Lock()
	w := f.w
	f.mu.Unlock()
	return w.Write(p)
}

func (f *framed) setMax(n uint32) { f.max.Store(n) }

var errTLSTimeout = errors.New("tls timeout")

func (s *Station) startTLS(ctx context.Context, f *framed, cfg *tls.Config, until time.Time) error {
	select {
	case f.upgrade <- cfg:
	case <-ctx.Done():
		return ctx.Err()
	}
	_ = f.conn.SetReadDeadline(time.Now())
	wait := until.Sub(s.clock.Now())
	if wait < 0 {
		wait = 0
	}
	timer := s.clock.NewTimer(wait)
	defer timer.Stop()
	select {
	case err := <-f.up:
		return err
	case <-ctx.Done():
		_ = f.conn.SetDeadline(time.Now())
		return ctx.Err()
	case <-timer.C():
		_ = f.conn.SetDeadline(time.Now())
		select {
		case <-f.up:
		case <-ctx.Done():
		}
		return errTLSTimeout
	}
}

func (f *framed) read(done <-chan struct{}) {
	src := io.Reader(f.conn)
	for {
		fr, err := readFrameMax(src, f.max.Load())
		if err != nil {
			var ne net.Error
			if errors.As(err, &ne) && ne.Timeout() {
				select {
				case cfg := <-f.upgrade:
					_ = f.conn.SetReadDeadline(time.Time{})
					tc := tls.Client(f.conn, cfg)
					herr := tc.Handshake()
					if herr != nil {
						f.up <- herr
						return
					}
					f.mu.Lock()
					f.w = tc
					f.mu.Unlock()
					src = tc
					f.up <- nil
					continue
				case <-done:
					return
				}
			}
			select {
			case f.errs <- err:
			case <-done:
			}
			return
		}
		select {
		case f.frames <- fr:
		case <-done:
			return
		}
	}
}

type phase int

const (
	phCA phase = iota
	phBundle
	phStart
	phIdent
	phServer
	phUDP
	phBeat
	phRun
	phChallenge
)

func (s *Station) session(ctx context.Context, conn net.Conn, major uint16) 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:
		}
	}()

	limit := uint32(maxFrame)
	ph := phCA
	label := "3.0"
	if major < 3 {
		limit = maxPreauth20
		ph = phChallenge
		label = "2.0"
	}
	wire := newFramed(conn, limit, done)
	s.mu.Lock()
	s.protocol = label
	s.rxSeq = 0
	s.rxCounter = 0
	s.txCounter = 1
	s.secure = false
	s.udpWait = false
	s.seq = 0
	s.ivPrefix = nil
	s.udpKey = nil
	s.mu.Unlock()

	if err := writeFrame(wire, tcpProtoVer, versionWire(major, 0)); err != nil {
		return err
	}
	now := s.clock.Now()
	lastTCP := now
	tcpRx := now.Add(15 * time.Second)
	udpRx := now.Add(60 * time.Second)
	authed := false

	var udpCh chan []byte
	var udpErr chan error
	udpReady := false
	var client tls.Certificate
	var haveCert bool

	for {
		s.mu.Lock()
		lastUDP := s.lastUDPSend
		silence, silenceOK := s.nextSilence()
		s.mu.Unlock()

		next := lastTCP.Add(10 * time.Second)
		if t := tcpRx; t.Before(next) {
			next = t
		}
		if t := udpRx; t.Before(next) {
			next = t
		}
		if udpReady {
			if t := lastUDP.Add(15 * time.Second); t.Before(next) {
				next = t
			}
		}
		if silenceOK && silence.Before(next) {
			next = silence
		}
		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 err := <-wire.errs:
			timer.Stop()
			return err
		case err := <-udpFail:
			timer.Stop()
			return err
		case tcpf = <-wire.frames:
			timer.Stop()
			reason = "tcp"
		case udpp = <-udpRecv:
			timer.Stop()
			reason = "udp"
		case <-timer.C():
			select {
			case tcpf = <-wire.frames:
				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)
			limit := tcpRx
			if udpRx.Before(limit) {
				limit = udpRx
			}
			err := s.onFrame(ctx, wire, major, tcpf, &ph, &authed, &client, &haveCert, done, &udpCh, &udpErr, &udpReady, &lastTCP, limit)
			if err != nil {
				return err
			}
		case "udp":
			if err := s.onUDP(wire, udpp, &ph, &udpRx, now, &lastTCP); err != nil {
				return err
			}
		default:
			s.mu.Lock()
			s.onClock(now)
			s.mu.Unlock()
			if !now.Before(tcpRx) {
				return s.retry("tcp receive timeout")
			}
			if !now.Before(udpRx) {
				return s.retry("udp receive timeout")
			}
			if !now.Before(lastTCP.Add(10 * time.Second)) {
				if err := writeFrame(wire, 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)) {
				var err error
				if s.udpWait {
					err = s.sendRegistration()
				} else {
					err = s.sendUDP(udpHeartbeat, nil)
				}
				if err != nil {
					s.mu.Unlock()
					return err
				}
			}
			s.mu.Unlock()
		}
	}
}

func (s *Station) retry(reason string) error {
	s.mu.Lock()
	s.retryReason = reason
	s.mu.Unlock()
	return errors.New(reason)
}

func (s *Station) nextSilence() (time.Time, bool) {
	var next time.Time
	ok := false
	if s.audioHeard {
		next = s.lastAudio.Add(3*time.Second + time.Nanosecond)
		ok = true
	}
	if s.flushWaiting {
		at := s.flushAt.Add(3 * time.Second)
		if !ok || at.Before(next) {
			next = at
			ok = true
		}
	}
	return next, ok
}

func (s *Station) onClock(now time.Time) {
	if s.flushWaiting && !now.Before(s.flushAt.Add(3*time.Second)) {
		s.flushWaiting = false
	}
	if s.audioHeard && now.After(s.lastAudio.Add(3*time.Second)) {
		if s.dec != nil {
			s.dec.Reset()
		}
		s.audioHeard = false
		s.roomBusy = false
		s.publish()
	}
}

func (s *Station) onFrame(ctx context.Context, wire *framed, major uint16, f frame, ph *phase, authed *bool, client *tls.Certificate, haveCert *bool, done chan struct{}, udpCh *chan []byte, udpErr *chan error, udpReady *bool, lastTCP *time.Time, limit time.Time) error {
	switch f.typ {
	case tcpHeartbeat:
		return nil
	case tcpError:
		return s.onReflectorError(f.body, major)
	case tcpProtoDown:
		return s.onDowngrade(f.body, major)
	}
	if !*authed && f.typ >= 100 {
		return s.retry("Protocol error")
	}
	if major < 3 {
		return s.onV2(wire, f, ph, authed, done, udpCh, udpErr, udpReady, lastTCP)
	}
	return s.onV3(ctx, wire, f, ph, authed, client, haveCert, done, udpCh, udpErr, udpReady, lastTCP, limit)
}

func (s *Station) onReflectorError(body []byte, major uint16) error {
	msg, _, err := parseString(body)
	if err != nil {
		msg = "could not be read"
	}
	if major >= 3 && fallback20(msg) {
		return errFallback
	}
	s.endFatal(msg, true)
	return errStopped
}

func (s *Station) onDowngrade(body []byte, major uint16) error {
	gotMajor, gotMinor, err := parseVersion(body)
	if err != nil {
		return err
	}
	if major >= 3 && gotMajor == 2 {
		return errFallback
	}
	s.endFatal("Unsupported protocol version "+strconvU(gotMajor)+"."+strconvU(gotMinor), true)
	return errStopped
}

func strconvU(v uint16) string {
	return strconv.FormatUint(uint64(v), 10)
}

func (s *Station) onV2(wire *framed, f frame, ph *phase, authed *bool, done chan struct{}, udpCh *chan []byte, udpErr *chan error, udpReady *bool, lastTCP *time.Time) error {
	switch f.typ {
	case tcpAuthOK:
		*authed = true
		*ph = phServer
		wire.setMax(maxFrame)
		return nil
	case tcpAuthChallenge:
		return s.answerChallenge(wire, f.body, lastTCP)
	case tcpServerInfo:
		if *ph != phServer {
			return s.retry("Protocol error")
		}
		return s.finishV2(wire, f.body, done, udpCh, udpErr, udpReady, lastTCP)
	case tcpTalkerStart:
		s.onTalker(f.body, true)
		return nil
	case tcpTalkerStop:
		s.onTalker(f.body, false)
		return nil
	case tcpNodeList, tcpNodeJoined, tcpNodeLeft, tcpRequestQSY, tcpStateEvent, tcpSignal, tcpTxStatus:
		if err := unpackIgnored(f); err != nil {
			return s.retry("Protocol error")
		}
		return nil
	default:
		return nil
	}
}

func (s *Station) answerChallenge(w io.Writer, body []byte, lastTCP *time.Time) error {
	chal, _, err := parseBlob(body)
	if err != nil || len(chal) != 20 {
		return errors.New("bad challenge")
	}
	mac := hmac.New(sha1.New, []byte(s.cfg.authKey))
	mac.Write(chal)
	out := appendString(nil, s.cfg.callsign)
	out = appendBlob(out, mac.Sum(nil))
	if err := writeFrame(w, tcpAuthResponse, out); err != nil {
		return err
	}
	*lastTCP = s.clock.Now()
	return nil
}

func (s *Station) onV3(ctx context.Context, wire *framed, f frame, ph *phase, authed *bool, client *tls.Certificate, haveCert *bool, done chan struct{}, udpCh *chan []byte, udpErr *chan error, udpReady *bool, lastTCP *time.Time, limit time.Time) error {
	switch f.typ {
	case tcpCAInfo:
		if *ph != phCA {
			return nil
		}
		return s.onCAInfo(wire, f.body, ph, lastTCP)
	case tcpCABundle:
		if *ph != phBundle {
			return nil
		}
		return s.onCABundle(wire, f.body, ph, lastTCP)
	case tcpStartEnc:
		if *ph != phStart {
			return nil
		}
		return s.onStartEncryption(ctx, wire, ph, client, haveCert, limit)
	case tcpCsrReq:
		return s.sendCSR(wire, lastTCP)
	case tcpAuthChallenge:
		return s.answerChallenge(wire, f.body, lastTCP)
	case tcpAuthOK:
		*authed = true
		*ph = phServer
		if csrNewerThanCert(s.cfg.certificateDir) {
			if err := s.sendCSR(wire, lastTCP); err != nil {
				return err
			}
		}
		return nil
	case tcpClientCert:
		return s.saveCert(f.body)
	case tcpServerInfo:
		if !*authed {
			return s.retry("Protocol error")
		}
		return s.finishV3Server(wire, f.body, ph, lastTCP)
	case tcpStartUDPEnc:
		if *ph != phUDP {
			return nil
		}
		return s.finishV3UDP(wire.conn, done, udpCh, udpErr, udpReady, ph)
	case tcpTalkerStart:
		s.onTalker(f.body, true)
		return nil
	case tcpTalkerStop:
		s.onTalker(f.body, false)
		return nil
	case tcpNodeList, tcpNodeJoined, tcpNodeLeft, tcpRequestQSY, tcpStateEvent, tcpSignal, tcpTxStatus:
		if err := unpackIgnored(f); err != nil {
			return s.retry("Protocol error")
		}
		return nil
	default:
		return nil
	}
}

func (s *Station) onCAInfo(w io.Writer, body []byte, ph *phase, lastTCP *time.Time) error {
	_, digest, err := parseCAInfo(body)
	if err != nil {
		return s.retry("Protocol error")
	}
	if err := ensureCertDir(s.cfg.certificateDir); err != nil {
		s.endFatal("Cannot create "+s.cfg.certificateDir, true)
		return errStopped
	}
	sum, ok, err := caDigest(s.cfg.certificateDir)
	if err != nil {
		s.endFatal("Cannot create "+s.cfg.certificateDir, true)
		return errStopped
	}
	if ok {
		if !bytes.Equal(sum[:], digest) {
			s.mu.Lock()
			s.writeLog(s.clock.Now(), "CA digest differs")
			s.mu.Unlock()
		}
		if err := writeFrame(w, tcpStartEncReq, nil); err != nil {
			return err
		}
		*lastTCP = s.clock.Now()
		*ph = phStart
		return nil
	}
	if err := writeFrame(w, tcpCABundleReq, nil); err != nil {
		return err
	}
	*lastTCP = s.clock.Now()
	*ph = phBundle
	return nil
}

func parseCAInfo(b []byte) (uint16, []byte, error) {
	size, rest, err := parseU16(b)
	if err != nil {
		return 0, nil, err
	}
	digest, rest, err := parseBlob(rest)
	if err != nil {
		return 0, nil, err
	}
	if len(digest) != sha256Len || len(rest) != 0 {
		return 0, nil, io.ErrUnexpectedEOF
	}
	return size, digest, nil
}

const sha256Len = 32

func (s *Station) onCABundle(w io.Writer, body []byte, ph *phase, lastTCP *time.Time) error {
	caPEM, rest, err := parseString(body)
	if err != nil {
		return s.retry("CA signature rejected")
	}
	sig, rest, err := parseBlob(rest)
	if err != nil {
		return s.retry("CA signature rejected")
	}
	certPEM, rest, err := parseString(rest)
	if err != nil || len(rest) != 0 {
		return s.retry("CA signature rejected")
	}
	if err := verifyCABundle(caPEM, sig, certPEM); err != nil {
		return s.retry("CA signature rejected")
	}
	if err := writeCABundle(s.cfg.certificateDir, caPEM); err != nil {
		s.endFatal("Cannot create "+s.cfg.certificateDir, true)
		return errStopped
	}
	if err := writeFrame(w, tcpStartEncReq, nil); err != nil {
		return err
	}
	*lastTCP = s.clock.Now()
	*ph = phStart
	return nil
}

func (s *Station) onStartEncryption(ctx context.Context, wire *framed, ph *phase, client *tls.Certificate, haveCert *bool, limit time.Time) error {
	pair, present, fatal := prepareIdentity(s.cfg.certificateDir, s.cfg.callsign)
	if fatal != "" {
		s.endFatal(fatal, true)
		return errStopped
	}
	*client = pair
	*haveCert = present
	caPEM, err := os.ReadFile(filepath.Join(s.cfg.certificateDir, caFileName))
	if err != nil {
		s.endFatal("Cannot create "+s.cfg.certificateDir, true)
		return errStopped
	}
	cfg, err := tlsConfig(s.cfg.host, caPEM, pair, present)
	if err != nil {
		return s.retry("TLS handshake failed")
	}
	if err := s.startTLS(ctx, wire, cfg, limit); err != nil {
		if errors.Is(err, errHostMismatch) {
			return s.retry("Reflector certificate does not match the host")
		}
		if ctx.Err() != nil {
			return ctx.Err()
		}
		return s.retry("TLS handshake failed")
	}
	*ph = phIdent
	return nil
}

func (s *Station) sendCSR(w io.Writer, lastTCP *time.Time) error {
	pem, err := readCSR(s.cfg.certificateDir)
	if err != nil {
		s.endFatal("Cannot create "+s.cfg.certificateDir, true)
		return errStopped
	}
	if err := writeFrame(w, tcpCsr, appendString(nil, pem)); err != nil {
		return err
	}
	*lastTCP = s.clock.Now()
	return nil
}

func (s *Station) saveCert(body []byte) error {
	pem, _, err := parseString(body)
	if err != nil || pem == "" {
		return s.retry("Protocol error")
	}
	if err := writeClientCert(s.cfg.certificateDir, pem); err != nil {
		s.endFatal("Cannot create "+s.cfg.certificateDir, true)
		return errStopped
	}
	return errReconnect
}

func (s *Station) finishV2(wire *framed, body []byte, done chan struct{}, udpCh *chan []byte, udpErr *chan error, udpReady *bool, lastTCP *time.Time) error {
	id, enc, dec, err := s.acceptServer(body)
	if err != nil {
		return err
	}
	if enc == nil {
		return errStopped
	}
	info, err := softwareIdentityBody()
	if err != nil {
		enc.Close()
		dec.Close()
		return err
	}
	if err := writeFrame(wire, tcpNodeInfo, info); err != nil {
		enc.Close()
		dec.Close()
		return err
	}
	var tg [4]byte
	binary.BigEndian.PutUint32(tg[:], uint32(s.cfg.talkGroup))
	if err := writeFrame(wire, tcpSelectTG, tg[:]); err != nil {
		enc.Close()
		dec.Close()
		return err
	}
	*lastTCP = s.clock.Now()
	udpConn, peer, err := dialAudio(wire.conn)
	if err != nil {
		enc.Close()
		dec.Close()
		return err
	}
	ch := make(chan []byte, 32)
	errc := make(chan error, 1)
	go readUDP(udpConn, peer, ch, errc, done)
	s.mu.Lock()
	s.clientID = id
	s.udpPeer = peer
	s.seq = 0
	s.rxSeq = 0
	s.udp = udpConn
	s.enc = enc
	s.dec = dec
	s.secure = false
	err = s.sendUDP(udpHeartbeat, nil)
	if err != nil {
		s.mu.Unlock()
		enc.Close()
		dec.Close()
		udpConn.Close()
		return err
	}
	s.state.Connection = "Connected"
	s.up = true
	s.outage = false
	s.noteTransmitLocked()
	s.writeLog(s.clock.Now(), "connected")
	s.publish()
	s.mu.Unlock()
	*udpCh = ch
	*udpErr = errc
	*udpReady = true
	return nil
}

func (s *Station) acceptServer(body []byte) (uint16, *opus.Encoder, *opus.Decoder, error) {
	id, codecs, err := parseServerInfo(body)
	if err != nil {
		return 0, nil, nil, err
	}
	if !hasOPUS(codecs) {
		s.endFatal(noOpus, true)
		return 0, nil, nil, errStopped
	}
	enc, err := opus.NewEncoder()
	if err != nil {
		return 0, nil, nil, err
	}
	dec, err := opus.NewDecoder()
	if err != nil {
		enc.Close()
		return 0, nil, nil, err
	}
	return id, enc, dec, nil
}

func (s *Station) finishV3Server(wire *framed, body []byte, ph *phase, lastTCP *time.Time) error {
	id, enc, dec, err := s.acceptServer(body)
	if err != nil {
		return err
	}
	if enc == nil {
		return errStopped
	}
	iv := make([]byte, 6)
	key := make([]byte, 16)
	if _, err := rand.Read(iv); err != nil || randFill(key) != nil {
		enc.Close()
		dec.Close()
		return errors.New("udp key")
	}
	info, err := nodeInfoV3(iv, key)
	if err != nil {
		enc.Close()
		dec.Close()
		return err
	}
	if err := writeFrame(wire, tcpNodeInfo, info); err != nil {
		enc.Close()
		dec.Close()
		return err
	}
	*lastTCP = s.clock.Now()
	s.mu.Lock()
	s.clientID = id
	s.enc = enc
	s.dec = dec
	s.ivPrefix = iv
	s.udpKey = key
	s.secure = true
	s.mu.Unlock()
	*ph = phUDP
	return nil
}

func randFill(b []byte) error {
	_, err := rand.Read(b)
	return err
}

func (s *Station) finishV3UDP(conn net.Conn, done chan struct{}, udpCh *chan []byte, udpErr *chan error, udpReady *bool, ph *phase) error {
	udpConn, peer, err := dialAudio(conn)
	if err != nil {
		s.dropAudio()
		return err
	}
	ch := make(chan []byte, 32)
	errc := make(chan error, 1)
	go readUDP(udpConn, peer, ch, errc, done)
	s.mu.Lock()
	s.udp = udpConn
	s.udpPeer = peer
	s.udpWait = true
	s.txCounter = 1
	s.rxCounter = 0
	err = s.sendRegistration()
	s.mu.Unlock()
	if err != nil {
		s.dropAudio()
		udpConn.Close()
		return err
	}
	*udpCh = ch
	*udpErr = errc
	*udpReady = true
	*ph = phBeat
	return nil
}

func dialAudio(conn net.Conn) (*net.UDPConn, *net.UDPAddr, error) {
	remote, ok := conn.RemoteAddr().(*net.TCPAddr)
	if !ok {
		return nil, nil, errors.New("tcp peer has no address")
	}
	local := &net.UDPAddr{Port: 0}
	if tcpLocal, ok := conn.LocalAddr().(*net.TCPAddr); ok && tcpLocal.IP != nil {
		local.IP = tcpLocal.IP
	}
	udpConn, err := net.ListenUDP("udp", local)
	if err != nil {
		return nil, nil, err
	}
	return udpConn, &net.UDPAddr{IP: append(net.IP(nil), remote.IP...), Port: remote.Port}, nil
}

func (s *Station) dropAudio() {
	s.mu.Lock()
	defer s.mu.Unlock()
	if s.enc != nil {
		s.enc.Close()
		s.enc = nil
	}
	if s.dec != nil {
		s.dec.Close()
		s.dec = nil
	}
}

func readUDP(conn *net.UDPConn, peer *net.UDPAddr, out chan<- []byte, errc chan<- error, done <-chan struct{}) {
	buf := make([]byte, 2048)
	for {
		n, addr, err := conn.ReadFromUDP(buf)
		if err != nil {
			select {
			case errc <- err:
			case <-done:
			}
			return
		}
		if addr == nil || peer == nil || !addr.IP.Equal(peer.IP) || addr.Port != peer.Port {
			continue
		}
		pkt := append([]byte(nil), buf[:n]...)
		select {
		case out <- pkt:
		case <-done:
			return
		}
	}
}

func (s *Station) onUDP(w io.Writer, pkt []byte, ph *phase, udpRx *time.Time, now time.Time, lastTCP *time.Time) error {
	s.mu.Lock()
	defer s.mu.Unlock()
	if s.secure {
		plain, ok := s.openSecure(pkt)
		if !ok {
			return nil
		}
		*udpRx = now.Add(60 * time.Second)
		if len(plain) < 2 {
			return nil
		}
		typ := binary.BigEndian.Uint16(plain[:2])
		body := plain[2:]
		if s.udpWait {
			if typ != udpHeartbeat {
				return nil
			}
			s.udpWait = false
			s.mu.Unlock()
			var tg [4]byte
			binary.BigEndian.PutUint32(tg[:], uint32(s.cfg.talkGroup))
			err := writeFrame(w, tcpSelectTG, tg[:])
			s.mu.Lock()
			if err != nil {
				return err
			}
			*lastTCP = s.clock.Now()
			s.state.Connection = "Connected"
			s.up = true
			s.outage = false
			s.noteTransmitLocked()
			s.writeLog(s.clock.Now(), "connected")
			s.publish()
			*ph = phRun
			return nil
		}
		s.onPlain(typ, body, now)
		return nil
	}
	if !s.acceptV2(pkt) {
		return nil
	}
	*udpRx = now.Add(60 * time.Second)
	typ := binary.BigEndian.Uint16(pkt[:2])
	s.onPlain(typ, pkt[6:], now)
	return nil
}

func (s *Station) openSecure(pkt []byte) ([]byte, bool) {
	if len(pkt) < 12 || s.udpKey == nil {
		return nil, false
	}
	counter := binary.BigEndian.Uint32(pkt[:4])
	if counter < s.rxCounter {
		return nil, false
	}
	tag := pkt[4:12]
	ct := pkt[12:]
	plain, err := openDatagram(s.udpKey, s.ivPrefix, 0, counter, pkt[:4], tag, ct)
	if err != nil {
		return nil, false
	}
	s.rxCounter = counter + 1
	return plain, true
}

func (s *Station) acceptV2(pkt []byte) bool {
	if len(pkt) < 6 {
		return false
	}
	if binary.BigEndian.Uint16(pkt[2:4]) != s.clientID {
		return false
	}
	seq := binary.BigEndian.Uint16(pkt[4:6])
	diff := seq - s.rxSeq
	if diff > 32767 {
		return false
	}
	s.rxSeq = seq + 1
	return true
}

func (s *Station) onPlain(typ uint16, body []byte, now time.Time) {
	switch typ {
	case udpAllFlushed:
		s.flushWaiting = false
	case udpAudio:
		if s.dec == nil || s.transmitting() {
			return
		}
		packet, _, err := parseBlob(body)
		if err != nil {
			return
		}
		pcm, err := s.dec.Decode(packet)
		if err != nil {
			return
		}
		s.play = append(s.play, pcm...)
		s.roomBusy = true
		s.audioHeard = true
		s.lastAudio = now
		s.publish()
	}
}

func (s *Station) sendRegistration() error {
	return s.sendSecure(udpHeartbeat, nil, 0, true)
}

func (s *Station) sendUDP(typ uint16, body []byte) error {
	if s.udp == nil {
		return errors.New("udp is down")
	}
	if s.secure {
		return s.sendSecure(typ, body, s.txCounter, false)
	}
	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.WriteToUDP(buf, s.udpPeer); err != nil {
		return err
	}
	s.lastUDPSend = s.clock.Now()
	return nil
}

func (s *Station) sendSecure(typ uint16, body []byte, counter uint32, reg bool) error {
	plain := appendU16(nil, typ)
	plain = append(plain, body...)
	var aad []byte
	id := s.clientID
	if reg {
		aad = registrationAAD(s.clientID)
		counter = 0
	} else {
		aad = counterAAD(counter)
	}
	wire, err := sealDatagram(s.udpKey, s.ivPrefix, id, counter, aad, plain)
	if err != nil {
		return err
	}
	if _, err := s.udp.WriteToUDP(wire, s.udpPeer); err != nil {
		return err
	}
	s.lastUDPSend = s.clock.Now()
	if !reg {
		s.txCounter = counter + 1
	}
	return nil
}

func (s *Station) stop(msg string) { s.endFatal(msg, false) }

// 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()
	now := s.clock.Now()
	if start {
		if call == s.cfg.callsign {
			return
		}
		s.closeTalkerLocked(now)
		was := s.transmitting()
		s.remoteTalker = call
		s.roomBusy = true
		s.talkCall = call
		s.talkStart = now
		if was && !s.transmitting() {
			s.endTransmit()
		}
		s.publish()
		return
	}
	s.closeTalkerLocked(now)
	s.remoteTalker = ""
	s.roomBusy = false
	s.publish()
}

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)
		s.flushWaiting = true
		s.flushAt = s.clock.Now()
	}
}

func (s *Station) closeUDP() {
	if s.udp != nil {
		s.udp.Close()
		s.udp = nil
	}
	s.udpPeer = 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.noteTransmitLocked()
	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
		s.state.Protocol = ""
	} else {
		tx := s.transmitting()
		s.state.Transmitting = tx
		if tx {
			s.state.Talker = s.cfg.callsign
			s.play = nil
		} else {
			s.state.Talker = s.remoteTalker
		}
		if s.state.Connection == "Connected" {
			s.state.Protocol = s.protocol
			s.retryReason = ""
			s.state.Detail = "Idle"
			if s.state.Talker == "" && s.logNotice != "" {
				s.state.Detail = s.logNotice
			}
		} else {
			s.state.Protocol = ""
			if s.retryReason != "" {
				s.state.Detail = s.retryReason
			} else if s.state.Talker == "" && s.logNotice != "" {
				s.state.Detail = s.logNotice
			} else {
				s.state.Detail = "Idle"
			}
		}
	}
	select {
	case s.wake <- struct{}{}:
	default:
	}
}

func unpackIgnored(f frame) error {
	switch f.typ {
	case tcpNodeList:
		_, _, err := parseStringList(f.body)
		return err
	case tcpNodeJoined, tcpNodeLeft:
		_, _, err := parseString(f.body)
		return err
	case tcpRequestQSY:
		_, _, err := parseU32(f.body)
		return err
	case tcpStateEvent:
		_, rest, err := parseString(f.body)
		if err != nil {
			return err
		}
		_, rest, err = parseString(rest)
		if err != nil {
			return err
		}
		_, _, err = parseString(rest)
		return err
	case tcpSignal:
		return unpackRecords(f.body, 3)
	case tcpTxStatus:
		return unpackRecords(f.body, 2)
	default:
		return nil
	}
}

func unpackRecords(b []byte, size int) error {
	n, rest, err := parseU16(b)
	if err != nil {
		return err
	}
	if len(rest) != int(n)*size {
		return io.ErrUnexpectedEOF
	}
	return nil
}