funknetz/station/session.go

package station

import (
	"context"
	"crypto/hmac"
	"crypto/sha1"
	"crypto/tls"
	"encoding/binary"
	"errors"
	"fmt"
	"io"
	"net"
	"strconv"
	"strings"
	"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.loaded {
			s.mu.Unlock()
			select {
			case <-ctx.Done():
				return nil
			case <-s.wake:
			}
			continue
		}
		s.kick = false
		s.state.Connection = "Connecting"
		s.publish()
		host, port := s.cfg.host, s.cfg.port
		epoch := s.epoch
		dialCtx, dialCancel := context.WithCancel(ctx)
		s.cancelDial = dialCancel
		s.mu.Unlock()

		conn, err := dialStation(dialCtx, host, port)
		dialCancel()
		s.mu.Lock()
		s.cancelDial = nil
		stale := s.epoch != epoch
		s.mu.Unlock()
		if stale {
			if conn != nil {
				// This dial used the config from before Save.
				_ = conn.Close()
			}
			s.drainApply()
			continue
		}
		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, epoch)
		conn.Close()
		fallback := errors.Is(err, errFallback)
		reconnect := errors.Is(err, errReconnect)
		s.mu.Lock()
		s.closeUDP()
		if s.conn == conn {
			s.conn = nil
		}
		s.mic = nil
		s.play = nil
		s.audioHeard = false
		s.flushWaiting = false
		kicked := s.kick
		if kicked {
			s.kick = false
			s.drainApply()
			if s.up {
				s.noteDownLocked(s.clock.Now(), false)
			}
		}
		if fallback || reconnect || kicked {
			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
	case <-s.apply:
		return true
	}
}

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

// exchange is the state one TCP session's frame handlers share.
type exchange struct {
	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
}

func newFramed(conn net.Conn, max uint32) *framed {
	f := &framed{
		conn:      conn,
		w:         conn,
		frames:    make(chan frame, 8),
		errs:      make(chan error, 1),
		upgrade:   make(chan *tls.Config, 1),
		handshake: make(chan error, 1),
	}
	f.max.Store(max)
	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()
	}
	if err := f.conn.SetReadDeadline(time.Now()); err != nil {
		return err
	}
	wait := until.Sub(s.clock.Now())
	if wait < 0 {
		wait = 0
	}
	timer := s.clock.NewTimer(wait)
	defer timer.Stop()
	select {
	case err := <-f.handshake:
		return err
	case <-ctx.Done():
		if err := f.conn.SetDeadline(time.Now()); err != nil {
			return err
		}
		return ctx.Err()
	case <-timer.C():
		if err := f.conn.SetDeadline(time.Now()); err != nil {
			return err
		}
		select {
		case <-f.handshake:
		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 {
			if ne, ok := errors.AsType[net.Error](err); ok && ne.Timeout() {
				select {
				case cfg := <-f.upgrade:
					if derr := f.conn.SetReadDeadline(time.Time{}); derr != nil {
						f.handshake <- derr
						return
					}
					tc := tls.Client(f.conn, cfg)
					herr := tc.Handshake()
					if herr != nil {
						f.handshake <- herr
						return
					}
					f.mu.Lock()
					f.w = tc
					f.mu.Unlock()
					src = tc
					f.handshake <- 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, epoch uint64) error {
	s.mu.Lock()
	s.conn = conn
	stale := s.epoch != epoch
	s.mu.Unlock()
	if stale {
		return errReconnect
	}
	defer func() {
		s.mu.Lock()
		if s.conn == conn {
			s.conn = nil
		}
		s.mu.Unlock()
	}()
	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:
		}
	}()

	frameMax := uint32(maxFrame)
	ex := &exchange{ph: phCA, done: done, lastTCP: s.clock.Now()}
	label := "3.0"
	if major < 3 {
		frameMax = maxPreauth20
		ex.ph = phChallenge
		label = "2.0"
	}
	wire := newFramed(conn, frameMax)
	go wire.read(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 := ex.lastTCP
	tcpRx := now.Add(15 * time.Second)
	udpRx := now.Add(60 * time.Second)

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

		next := ex.lastTCP.Add(10 * time.Second)
		if t := tcpRx; t.Before(next) {
			next = t
		}
		if t := udpRx; t.Before(next) {
			next = t
		}
		if ex.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 ex.udpCh != nil {
			udpRecv = ex.udpCh
			udpFail = ex.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)
			ex.limit = tcpRx
			if udpRx.Before(ex.limit) {
				ex.limit = udpRx
			}
			err := s.onFrame(ctx, wire, major, tcpf, ex)
			if err != nil {
				return err
			}
		case "udp":
			if err := s.onUDP(wire, udpp, ex, &udpRx, now); err != nil {
				return err
			}
		default:
			s.mu.Lock()
			s.onClock(now)
			s.mu.Unlock()
			if !now.Before(tcpRx) {
				return s.retry("tcp receive timeout", nil)
			}
			if !now.Before(udpRx) {
				return s.retry("udp receive timeout", nil)
			}
			if !now.Before(ex.lastTCP.Add(10 * time.Second)) {
				if err := writeFrame(wire, tcpHeartbeat, nil); err != nil {
					return err
				}
				ex.lastTCP = s.clock.Now()
			}
			s.mu.Lock()
			if ex.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(detail string, cause error) error {
	s.mu.Lock()
	s.retryReason = detail
	s.mu.Unlock()
	text := strings.ToLower(detail)
	if cause == nil {
		return errors.New(text)
	}
	return fmt.Errorf("%s: %w", text, cause)
}

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, ex *exchange) 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 !ex.authed && f.typ >= 100 {
		return s.retry("Protocol error", nil)
	}
	if major < 3 {
		return s.onV2(wire, f, ex)
	}
	return s.onV3(ctx, wire, f, ex)
}

func (s *Station) onReflectorError(body []byte, major uint16) error {
	msg, _, err := parseString(body)
	if err != nil {
		s.endFatalCause("could not be read", err, true)
		return fmt.Errorf("could not be read: %w", err)
	}
	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, ex *exchange) error {
	switch f.typ {
	case tcpAuthOK:
		ex.authed = true
		ex.ph = phServer
		wire.setMax(maxFrame)
		return nil
	case tcpAuthChallenge:
		return s.answerChallenge(wire, f.body, ex)
	case tcpServerInfo:
		if ex.ph != phServer {
			return s.retry("Protocol error", nil)
		}
		return s.finishV2(wire, f.body, ex)
	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", err)
		}
		return nil
	default:
		return nil
	}
}

func (s *Station) answerChallenge(w io.Writer, body []byte, ex *exchange) error {
	chal, _, err := parseBlob(body)
	if err != nil {
		return fmt.Errorf("bad challenge: %w", err)
	}
	if 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
	}
	ex.lastTCP = s.clock.Now()
	return nil
}

func (s *Station) finishV2(wire *framed, body []byte, ex *exchange) 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
	}
	ex.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, ex.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()
	ex.udpCh = ch
	ex.udpErr = errc
	ex.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) 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.configOpen || 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
	s.state.NeedsConfig = !s.loaded
	s.state.ConfigOpen = s.configOpen
	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"
			}
		}
	}
	if s.configOpen && s.configErr != "" {
		s.state.Detail = s.configErr
	} else if s.openBlocked != "" {
		s.state.Detail = s.openBlocked
	}
	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
}