diff --git a/CONTEXT.md b/CONTEXT.md
index 1dc484a..76aa0c5 100644
--- a/CONTEXT.md
+++ b/CONTEXT.md
@@ -29,8 +29,24 @@ The uppercase amateur call sign that names this station on the reflector.
_Avoid_: a lowercase spelling, username, node id
**Auth key**:
-The secret the network issues for this station’s callsign, used to prove identity to the reflector.
-_Avoid_: certificate, a password sent in the clear
+The shared secret the network issues for this station’s callsign. It proves the station on protocol 2.0, and on protocol 3.0 when the reflector has not signed a certificate yet.
+_Avoid_: a password sent in the clear, the certificate
+
+**Certificate**:
+An X.509 certificate whose common name is this station’s callsign. It proves the station on protocol 3.0.
+_Avoid_: auth key, a PEM pasted into the config file
+
+**Certificate directory**:
+The directory the station config names. It holds the CA bundle and this station’s key and certificate.
+_Avoid_: PEM text inside the config file
+
+**CA bundle**:
+The reflector’s certificate authority, kept in the certificate directory. Protocol 3.0 uses it to check the reflector.
+_Avoid_: the station certificate
+
+**Protocol**:
+The reflector session this station is speaking: 2.0 or 3.0.
+_Avoid_: TYPE, a config field the operator sets
**Push-to-talk**:
The operator holding the control that makes this station the talker. Releasing it returns the station to listening.
diff --git a/docs/adr/0004-ca-bundle-trust-on-first-use.md b/docs/adr/0004-ca-bundle-trust-on-first-use.md
new file mode 100644
index 0000000..a65e2f2
--- /dev/null
+++ b/docs/adr/0004-ca-bundle-trust-on-first-use.md
@@ -0,0 +1,3 @@
+# Trust the first CA bundle
+
+Protocol 3.0 checks the reflector with the CA bundle in the certificate directory. When that file is missing, the station stores the bundle after the signature check with the certificate inside the same message. A later digest that differs from the stored file leaves the file in place. Replacing a stored bundle automatically was rejected because the reflector could swap the trust anchor.
diff --git a/docs/adr/0005-protocol-3-with-explicit-2-fallback.md b/docs/adr/0005-protocol-3-with-explicit-2-fallback.md
new file mode 100644
index 0000000..9efdebf
--- /dev/null
+++ b/docs/adr/0005-protocol-3-with-explicit-2-fallback.md
@@ -0,0 +1,3 @@
+# Announce protocol 3.0, fall back only on an explicit 2.x maximum
+
+Every TCP connection announces protocol 3.0. The station opens one 2.0 session only when the reflector sends a downgrade, or an unsupported-version error, whose maximum is at least 2.0 and below 3.0. The connection after that 2.0 session ends announces 3.0 again. A maximum whose major is below 2 stops the station with the reflector’s message. A reflector `MsgError` stops the station until the operator quits. A bad CA signature, a TLS or hostname failure, a heartbeat timeout, and a dropped connection retry on 3.0, and the window keeps the reason between attempts. The Protocol line shows 2.0 or 3.0 while the session is up, so a plaintext session is visible. Staying on 3.0 after every failure was rejected because a reflector that still expects 2.0 would never accept the station. Falling back on any handshake failure was rejected because a TLS or CA problem would then send plaintext audio.
diff --git a/frontend/dist/app.css b/frontend/dist/app.css
index fe8dd09..bd688bc 100644
--- a/frontend/dist/app.css
+++ b/frontend/dist/app.css
@@ -14,6 +14,11 @@ main {
font-size: 14px;
}
+#protocol {
+ min-height: 1.4em;
+ margin: 0;
+}
+
#detail {
min-height: 1.4em;
font-size: 20px;
diff --git a/frontend/dist/app.js b/frontend/dist/app.js
index daff955..692b40e 100644
--- a/frontend/dist/app.js
+++ b/frontend/dist/app.js
@@ -21,6 +21,8 @@ function render(state) {
document.getElementById("connection").textContent = state.connection;
document.getElementById("callsign").textContent = state.callsign;
document.getElementById("talkgroup").textContent = String(state.talkGroup);
+ const protocol = document.getElementById("protocol");
+ protocol.textContent = state.protocol ? "Protocol " + state.protocol : "";
document.getElementById("detail").textContent = line(state);
button.classList.toggle("down", !!state.transmitting);
}
diff --git a/frontend/dist/index.html b/frontend/dist/index.html
index 0c0e279..307e3b9 100644
--- a/frontend/dist/index.html
+++ b/frontend/dist/index.html
@@ -10,6 +10,7 @@
Disconnected
Callsign
Talk group
+
Idle
Push to talk
diff --git a/internal/opus/opus.go b/internal/opus/opus.go
index 038f1eb..a17a224 100644
--- a/internal/opus/opus.go
+++ b/internal/opus/opus.go
@@ -5,6 +5,10 @@ package opus
#cgo pkg-config: opus
#include
+static void reset_decoder(OpusDecoder *dec) {
+ opus_decoder_ctl(dec, OPUS_RESET_STATE);
+}
+
static int configure_encoder(OpusEncoder *enc) {
int err;
err = opus_encoder_ctl(enc, OPUS_SET_BITRATE(20000));
@@ -114,6 +118,14 @@ func (d *Decoder) Decode(packet []byte) ([]int16, error) {
return pcm, nil
}
+// Reset clears decoder state after a gap in the received audio.
+func (d *Decoder) Reset() {
+ if d == nil || d.dec == nil {
+ return
+ }
+ C.reset_decoder(d.dec)
+}
+
// Close releases the decoder.
func (d *Decoder) Close() {
if d == nil || d.dec == nil {
diff --git a/main.go b/main.go
index e383678..c73d366 100644
--- a/main.go
+++ b/main.go
@@ -25,7 +25,7 @@ func main() {
err = wails.Run(&options.App{
Title: "funknetz",
Width: 420,
- Height: 280,
+ Height: 320,
BackgroundColour: &options.RGBA{R: 16, G: 20, B: 24, A: 255},
AssetServer: &assetserver.Options{Assets: assets},
OnStartup: app.startup,
diff --git a/station/behavior_test.go b/station/behavior_test.go
index 0c9e6f2..5d99692 100644
--- a/station/behavior_test.go
+++ b/station/behavior_test.go
@@ -45,17 +45,13 @@ func loginLog(t *testing.T, callsign, key string, tg int, id uint16, codecs []st
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)
- }
+ declineTo2(t, refl)
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)
+ typ, body := refl.readFrame(t)
if typ != 11 {
t.Fatalf("auth type %d", typ)
}
@@ -88,12 +84,41 @@ func loginLog(t *testing.T, callsign, key string, tg int, id uint16, codecs []st
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 {
+ 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}
}
+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"})
@@ -127,12 +152,15 @@ func TestDropRetriesUntilQuit(t *testing.T) {
}); 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 || body[1] != 2 {
- t.Fatalf("retry protocol type %d body %v", typ, body)
+ if typ != 5 || !versionIs(body, 3, 0) {
+ t.Fatalf("retry protocol type %d body %v, want 3.0", typ, body)
}
in.cancel()
in.clock.Advance(2 * time.Second)
@@ -240,7 +268,7 @@ func TestMissingOpusStopsRetries(t *testing.T) {
defer cancel()
go func() { _ = st.Run(ctx) }()
refl.accept(t)
- refl.readFrame(t)
+ declineTo2(t, refl)
challenge := make([]byte, 20)
refl.writeFrame(t, 10, appendBytes(nil, challenge))
refl.readFrame(t)
@@ -508,14 +536,15 @@ func expectAudio(t *testing.T, refl *reflector, dec *opus.Decoder, id, seq uint1
}
func tryFrame(r *reflector) (uint16, []byte, bool) {
- _ = r.tcp.SetReadDeadline(time.Now().Add(30 * time.Millisecond))
+ conn := r.conn()
+ _ = conn.SetReadDeadline(time.Now().Add(30 * time.Millisecond))
var hdr [4]byte
- if _, err := io.ReadFull(r.tcp, hdr[:]); err != nil {
+ 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(r.tcp, buf); err != nil {
+ if _, err := io.ReadFull(conn, buf); err != nil {
return 0, nil, false
}
if len(buf) < 2 {
diff --git a/station/cert.go b/station/cert.go
new file mode 100644
index 0000000..818c0a6
--- /dev/null
+++ b/station/cert.go
@@ -0,0 +1,279 @@
+package station
+
+import (
+ "crypto"
+ "crypto/rand"
+ "crypto/rsa"
+ "crypto/sha256"
+ "crypto/tls"
+ "crypto/x509"
+ "crypto/x509/pkix"
+ "encoding/asn1"
+ "encoding/pem"
+ "errors"
+ "io/fs"
+ "os"
+ "path/filepath"
+)
+
+const (
+ caFileName = "ca.pem"
+ keyFileName = "key.pem"
+ csrFileName = "csr.pem"
+ certFileName = "cert.pem"
+)
+
+var (
+ oidBasicConstraints = asn1.ObjectIdentifier{2, 5, 29, 19}
+ oidKeyUsage = asn1.ObjectIdentifier{2, 5, 29, 15}
+ oidExtKeyUsage = asn1.ObjectIdentifier{2, 5, 29, 37}
+ oidClientAuth = asn1.ObjectIdentifier{1, 3, 6, 1, 5, 5, 7, 3, 2}
+)
+
+func ensureCertDir(path string) error {
+ err := os.Mkdir(path, 0o700)
+ if err != nil {
+ if !errors.Is(err, fs.ErrExist) {
+ return err
+ }
+ info, statErr := os.Stat(path)
+ if statErr != nil {
+ return statErr
+ }
+ if !info.IsDir() {
+ return err
+ }
+ return nil
+ }
+ return os.Chmod(path, 0o700)
+}
+
+func caDigest(dir string) (sum [32]byte, ok bool, err error) {
+ body, err := os.ReadFile(filepath.Join(dir, caFileName))
+ if errors.Is(err, fs.ErrNotExist) {
+ return sum, false, nil
+ }
+ if err != nil {
+ return sum, false, err
+ }
+ return sha256.Sum256(body), true, nil
+}
+
+func verifyCABundle(caPEM string, sig []byte, certPEM string) error {
+ block, _ := pem.Decode([]byte(certPEM))
+ if block == nil {
+ return errors.New("signing certificate")
+ }
+ cert, err := x509.ParseCertificate(block.Bytes)
+ if err != nil {
+ return err
+ }
+ pub, ok := cert.PublicKey.(*rsa.PublicKey)
+ if !ok {
+ return errors.New("signing certificate")
+ }
+ sum := sha256.Sum256([]byte(caPEM))
+ return rsa.VerifyPKCS1v15(pub, crypto.SHA256, sum[:], sig)
+}
+
+func writeCABundle(dir, caPEM string) error {
+ return writeMode(filepath.Join(dir, caFileName), []byte(caPEM), 0o644)
+}
+
+func writeClientCert(dir, certPEM string) error {
+ return writeMode(filepath.Join(dir, certFileName), []byte(certPEM), 0o644)
+}
+
+func csrNewerThanCert(dir string) bool {
+ csr, err1 := os.Stat(filepath.Join(dir, csrFileName))
+ cert, err2 := os.Stat(filepath.Join(dir, certFileName))
+ if err1 != nil || err2 != nil {
+ return false
+ }
+ return csr.ModTime().After(cert.ModTime())
+}
+
+func readCSR(dir string) (string, error) {
+ body, err := os.ReadFile(filepath.Join(dir, csrFileName))
+ if err != nil {
+ return "", err
+ }
+ return string(body), nil
+}
+
+// prepareIdentity checks a saved certificate or creates the key and signing
+// request. fatal is a window message when the station must stop.
+func prepareIdentity(dir, callsign string) (tls.Certificate, bool, string) {
+ certPath := filepath.Join(dir, certFileName)
+ keyPath := filepath.Join(dir, keyFileName)
+ _, certErr := os.Stat(certPath)
+ _, keyErr := os.Stat(keyPath)
+ certExists := certErr == nil
+ keyExists := keyErr == nil
+ if certErr != nil && !errors.Is(certErr, fs.ErrNotExist) {
+ return tls.Certificate{}, false, "Cannot create " + certPath
+ }
+ if keyErr != nil && !errors.Is(keyErr, fs.ErrNotExist) {
+ return tls.Certificate{}, false, "Cannot create " + keyPath
+ }
+ if certExists && !keyExists {
+ return tls.Certificate{}, false, "Certificate key is missing"
+ }
+ if certExists && keyExists {
+ pair, msg := loadPair(certPath, keyPath, callsign)
+ if msg != "" {
+ return tls.Certificate{}, false, msg
+ }
+ return pair, true, ""
+ }
+ if msg := ensureKeyAndCSR(dir, callsign, !keyExists); msg != "" {
+ return tls.Certificate{}, false, msg
+ }
+ return tls.Certificate{}, false, ""
+}
+
+func loadPair(certPath, keyPath, callsign string) (tls.Certificate, string) {
+ cert, err := loadCert(certPath)
+ if err != nil {
+ return tls.Certificate{}, "Certificate does not match the key"
+ }
+ key, err := loadKey(keyPath)
+ if err != nil {
+ return tls.Certificate{}, "Certificate does not match the key"
+ }
+ pub, ok := cert.PublicKey.(*rsa.PublicKey)
+ if !ok || !pub.Equal(&key.PublicKey) {
+ return tls.Certificate{}, "Certificate does not match the key"
+ }
+ if cert.Subject.CommonName != callsign {
+ return tls.Certificate{}, "Certificate callsign does not match"
+ }
+ return tls.Certificate{
+ Certificate: [][]byte{cert.Raw},
+ PrivateKey: key,
+ Leaf: cert,
+ }, ""
+}
+
+func ensureKeyAndCSR(dir, callsign string, created bool) string {
+ keyPath := filepath.Join(dir, keyFileName)
+ var key *rsa.PrivateKey
+ var err error
+ if created {
+ key, err = rsa.GenerateKey(rand.Reader, 2048)
+ if err != nil {
+ return "Cannot create " + keyPath
+ }
+ der := x509.MarshalPKCS1PrivateKey(key)
+ body := pem.EncodeToMemory(&pem.Block{Type: "RSA PRIVATE KEY", Bytes: der})
+ if err := writeNew(keyPath, body, 0o600); err != nil {
+ return "Cannot create " + keyPath
+ }
+ } else {
+ key, err = loadKey(keyPath)
+ if err != nil {
+ return "Cannot create " + keyPath
+ }
+ }
+ csrPath := filepath.Join(dir, csrFileName)
+ _, csrErr := os.Stat(csrPath)
+ if created || errors.Is(csrErr, fs.ErrNotExist) {
+ if err := writeCSR(csrPath, key, callsign); err != nil {
+ return "Cannot create " + csrPath
+ }
+ }
+ return ""
+}
+
+func writeCSR(path string, key *rsa.PrivateKey, callsign string) error {
+ bc, err := asn1.Marshal(struct {
+ CA bool
+ }{CA: false})
+ if err != nil {
+ return err
+ }
+ // digitalSignature, keyEncipherment, keyAgreement: bits 0, 2, and 4.
+ ku, err := asn1.Marshal(asn1.BitString{Bytes: []byte{0xA8}, BitLength: 5})
+ if err != nil {
+ return err
+ }
+ eku, err := asn1.Marshal([]asn1.ObjectIdentifier{oidClientAuth})
+ if err != nil {
+ return err
+ }
+ der, err := x509.CreateCertificateRequest(rand.Reader, &x509.CertificateRequest{
+ Subject: pkix.Name{CommonName: callsign},
+ ExtraExtensions: []pkix.Extension{
+ {Id: oidBasicConstraints, Critical: true, Value: bc},
+ {Id: oidKeyUsage, Critical: true, Value: ku},
+ {Id: oidExtKeyUsage, Critical: false, Value: eku},
+ },
+ }, key)
+ if err != nil {
+ return err
+ }
+ body := pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE REQUEST", Bytes: der})
+ return writeMode(path, body, 0o644)
+}
+
+func loadCert(path string) (*x509.Certificate, error) {
+ body, err := os.ReadFile(path)
+ if err != nil {
+ return nil, err
+ }
+ block, _ := pem.Decode(body)
+ if block == nil {
+ return nil, errors.New("certificate")
+ }
+ return x509.ParseCertificate(block.Bytes)
+}
+
+func loadKey(path string) (*rsa.PrivateKey, error) {
+ body, err := os.ReadFile(path)
+ if err != nil {
+ return nil, err
+ }
+ block, _ := pem.Decode(body)
+ if block == nil {
+ return nil, errors.New("key")
+ }
+ switch block.Type {
+ case "RSA PRIVATE KEY":
+ return x509.ParsePKCS1PrivateKey(block.Bytes)
+ case "PRIVATE KEY":
+ k, err := x509.ParsePKCS8PrivateKey(block.Bytes)
+ if err != nil {
+ return nil, err
+ }
+ rk, ok := k.(*rsa.PrivateKey)
+ if !ok {
+ return nil, errors.New("key")
+ }
+ return rk, nil
+ default:
+ return nil, errors.New("key")
+ }
+}
+
+func writeNew(path string, body []byte, mode os.FileMode) error {
+ f, err := os.OpenFile(path, os.O_WRONLY|os.O_CREATE|os.O_EXCL, mode)
+ 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, mode)
+}
+
+func writeMode(path string, body []byte, mode os.FileMode) error {
+ if err := os.WriteFile(path, body, mode); err != nil {
+ return err
+ }
+ return os.Chmod(path, mode)
+}
diff --git a/station/cipher.go b/station/cipher.go
new file mode 100644
index 0000000..8c0eab4
--- /dev/null
+++ b/station/cipher.go
@@ -0,0 +1,85 @@
+package station
+
+import (
+ "crypto/aes"
+ "crypto/cipher"
+ "crypto/subtle"
+ "encoding/binary"
+ "errors"
+)
+
+func counterAAD(counter uint32) []byte {
+ var b [4]byte
+ binary.BigEndian.PutUint32(b[:], counter)
+ return b[:]
+}
+
+func registrationAAD(clientID uint16) []byte {
+ b := counterAAD(0)
+ return appendU16(b, clientID)
+}
+
+func ivFor(prefix []byte, clientID uint16, counter uint32) []byte {
+ iv := make([]byte, 12)
+ copy(iv, prefix)
+ binary.BigEndian.PutUint16(iv[6:8], clientID)
+ binary.BigEndian.PutUint32(iv[8:12], counter)
+ return iv
+}
+
+func sealDatagram(key, prefix []byte, clientID uint16, counter uint32, aad, plain []byte) ([]byte, error) {
+ tag, ct, err := sealTrunc(key, ivFor(prefix, clientID, counter), plain, aad)
+ if err != nil {
+ return nil, err
+ }
+ wire := make([]byte, 0, len(aad)+len(tag)+len(ct))
+ wire = append(wire, aad...)
+ wire = append(wire, tag...)
+ return append(wire, ct...), nil
+}
+
+func openDatagram(key, prefix []byte, clientID uint16, counter uint32, aad, tag, ct []byte) ([]byte, error) {
+ return openTrunc(key, ivFor(prefix, clientID, counter), tag, ct, aad)
+}
+
+// The protocol's tag is 8 bytes. crypto/cipher only builds GCM tags of 12 to 16
+// bytes, and an 8-byte tag is the leading 8 bytes of the ordinary 16-byte tag.
+func sealTrunc(key, nonce, plain, aad []byte) (tag, ct []byte, err error) {
+ g, err := gcmFull(key)
+ if err != nil {
+ return nil, nil, err
+ }
+ out := g.Seal(nil, nonce, plain, aad)
+ ct = append([]byte(nil), out[:len(plain)]...)
+ tag = append([]byte(nil), out[len(plain):len(plain)+8]...)
+ return tag, ct, nil
+}
+
+func openTrunc(key, nonce, tag, ct, aad []byte) ([]byte, error) {
+ if len(tag) != 8 {
+ return nil, errors.New("tag")
+ }
+ g, err := gcmFull(key)
+ if err != nil {
+ return nil, err
+ }
+ stream := g.Seal(nil, nonce, make([]byte, len(ct)), aad)
+ plain := make([]byte, len(ct))
+ subtle.XORBytes(plain, ct, stream[:len(ct)])
+ check := g.Seal(nil, nonce, plain, aad)
+ var got, want [8]byte
+ copy(got[:], tag)
+ copy(want[:], check[len(plain):len(plain)+8])
+ if subtle.ConstantTimeCompare(got[:], want[:]) != 1 {
+ return nil, errors.New("tag")
+ }
+ return plain, nil
+}
+
+func gcmFull(key []byte) (cipher.AEAD, error) {
+ block, err := aes.NewCipher(key)
+ if err != nil {
+ return nil, err
+ }
+ return cipher.NewGCM(block)
+}
diff --git a/station/config.go b/station/config.go
index 059826a..e86a272 100644
--- a/station/config.go
+++ b/station/config.go
@@ -15,34 +15,38 @@ import (
)
type loaded struct {
- host string
- port int
- callsign string
- authKey string
- talkGroup int
- enableLogging bool
- logPath string
- logNotice string
+ host string
+ port int
+ callsign string
+ authKey string
+ talkGroup int
+ certificateDir string
+ certFatal string
+ enableLogging bool
+ logPath string
+ logNotice string
}
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"`
- EnableLogging jsontext.Value `json:"enableLogging"`
- LogFile jsontext.Value `json:"logFile"`
+ 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"`
+ CertificateDir jsontext.Value `json:"certificateDir"`
+ EnableLogging jsontext.Value `json:"enableLogging"`
+ LogFile jsontext.Value `json:"logFile"`
}
type templateFile struct {
- Host string `json:"host"`
- Port int `json:"port"`
- Callsign string `json:"callsign"`
- AuthKey string `json:"authKey"`
- TalkGroup int `json:"talkGroup"`
- EnableLogging bool `json:"enableLogging"`
- LogFile string `json:"logFile"`
+ Host string `json:"host"`
+ Port int `json:"port"`
+ Callsign string `json:"callsign"`
+ AuthKey string `json:"authKey"`
+ TalkGroup int `json:"talkGroup"`
+ CertificateDir string `json:"certificateDir"`
+ EnableLogging bool `json:"enableLogging"`
+ LogFile string `json:"logFile"`
}
var callsignPattern = regexp.MustCompile(`^[A-Z0-9][A-Z]{0,2}\d[A-Z0-9]{0,3}[A-Z](?:-[A-Z0-9]{1,3})?$`)
@@ -66,7 +70,7 @@ func writeTemplate(path string) error {
if err != nil {
return err
}
- body, err := jsonv2.Marshal(templateFile{LogFile: logFile})
+ body, err := jsonv2.Marshal(templateFile{CertificateDir: "./funknetz", LogFile: logFile})
if err != nil {
return err
}
@@ -91,15 +95,19 @@ func parseConfig(path string, body []byte) (loaded, error) {
if err := jsonv2.Unmarshal(body, &shape); err != nil {
return loaded{}, fmt.Errorf("the file could not be read: %w", err)
}
- if block, notice := blockedLog(shape); block {
+ certBad := present(shape.CertificateDir) && shape.CertificateDir.Kind() != jsontext.KindString
+ if block, notice := blockedLog(shape); block || certBad {
cfg, err := parseOperational(shape)
if err != nil {
return loaded{}, err
}
cfg.logNotice = notice
- return cfg, nil
+ if !block {
+ cfg = applyLog(cfg, shape)
+ }
+ return applyCertificate(cfg, shape, path), nil
}
- if missingLog(shape) {
+ if missingCertificate(shape) || missingLog(shape) {
if !operationalKindsOK(shape) {
return parseOperational(shape)
}
@@ -116,7 +124,7 @@ func parseConfig(path string, body []byte) (loaded, error) {
if err != nil {
return loaded{}, err
}
- return applyLog(cfg, shape), nil
+ return applyLog(applyCertificate(cfg, shape, path), shape), nil
}
func parseOperational(shape fileShape) (loaded, error) {
@@ -217,6 +225,10 @@ func missingLog(shape fileShape) bool {
return !present(shape.EnableLogging) || !present(shape.LogFile)
}
+func missingCertificate(shape fileShape) bool {
+ return !present(shape.CertificateDir)
+}
+
func operationalKindsOK(shape fileShape) bool {
if present(shape.Host) && shape.Host.Kind() != jsontext.KindString {
return false
@@ -259,6 +271,7 @@ func withLogDefaults(shape fileShape) (fileShape, error) {
shape.Callsign = rawOr(shape.Callsign, `""`)
shape.AuthKey = rawOr(shape.AuthKey, `""`)
shape.TalkGroup = rawOr(shape.TalkGroup, "0")
+ shape.CertificateDir = rawOr(shape.CertificateDir, `"./funknetz"`)
shape.EnableLogging = rawOr(shape.EnableLogging, "false")
if !present(shape.LogFile) {
logFile, err := defaultLogPath()
@@ -316,6 +329,32 @@ func rewriteConfig(path string, shape fileShape) error {
return nil
}
+func applyCertificate(cfg loaded, shape fileShape, configPath string) loaded {
+ if !present(shape.CertificateDir) {
+ cfg.certificateDir = resolveCertDir(configPath, "./funknetz")
+ return cfg
+ }
+ dir, err := stringField("certificateDir", shape.CertificateDir)
+ if err != nil {
+ cfg.certFatal = err.Error()
+ return cfg
+ }
+ if strings.TrimSpace(dir) == "" {
+ cfg.certFatal = "certificateDir is empty"
+ return cfg
+ }
+ cfg.certificateDir = resolveCertDir(configPath, dir)
+ return cfg
+}
+
+func resolveCertDir(configPath, dir string) string {
+ dir = strings.TrimSpace(dir)
+ if filepath.IsAbs(dir) {
+ return filepath.Clean(dir)
+ }
+ return filepath.Clean(filepath.Join(filepath.Dir(configPath), dir))
+}
+
func applyLog(cfg loaded, shape fileShape) loaded {
if shape.EnableLogging.Kind() != jsontext.KindTrue {
return cfg
diff --git a/station/config_test.go b/station/config_test.go
index f40134c..0bfef63 100644
--- a/station/config_test.go
+++ b/station/config_test.go
@@ -1,10 +1,14 @@
package station_test
import (
+ "context"
+ "fmt"
+ "net"
"os"
"path/filepath"
"strings"
"testing"
+ "time"
jsonv2 "encoding/json/v2"
@@ -54,11 +58,15 @@ func TestMissingConfigIsCreatedAndRejected(t *testing.T) {
if got["logFile"] != filepath.Join(home, "funknetz.log") {
t.Fatalf("logFile = %#v, want funknetz.log in the home directory", got["logFile"])
}
+ if got["certificateDir"] != "./funknetz" {
+ t.Fatalf("certificateDir = %#v, want ./funknetz", got["certificateDir"])
+ }
+ assertKeyOrder(t, body)
}
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,\"enableLogging\":false,\"logFile\":\"/tmp/funknetz.log\"}\n")
+ original := []byte("{\"host\":\"127.0.0.1\",\"port\":5308,\"callsign\":\"DL1ABC\",\"authKey\":\"a b\",\"talkGroup\":20,\"certificateDir\":\"/var/funknetz\",\"enableLogging\":false,\"logFile\":\"/tmp/funknetz.log\"}\n")
if err := os.WriteFile(path, original, 0o640); err != nil {
t.Fatal(err)
}
@@ -240,6 +248,10 @@ func TestMissingLogKeysAreRewritten(t *testing.T) {
if got["logFile"] != filepath.Join(home, "funknetz.log") {
t.Fatalf("logFile = %#v", got["logFile"])
}
+ if got["certificateDir"] != "./funknetz" {
+ t.Fatalf("certificateDir = %#v, want ./funknetz", got["certificateDir"])
+ }
+ assertKeyOrder(t, body)
if st.State().Detail != "Idle" {
t.Fatalf("Detail = %q, want Idle", st.State().Detail)
}
@@ -314,9 +326,9 @@ func TestLogSettingsDoNotStopTheStation(t *testing.T) {
detail string
same bool
}{
- {name: "relative", body: `{"host":"127.0.0.1","port":5308,"callsign":"DL1ABC","authKey":"key","talkGroup":20,"enableLogging":true,"logFile":"funknetz.log"}`, detail: "Cannot write funknetz.log", same: true},
- {name: "empty", body: `{"host":"127.0.0.1","port":5308,"callsign":"DL1ABC","authKey":"key","talkGroup":20,"enableLogging":true,"logFile":" "}`, detail: "Cannot write logFile", same: true},
- {name: "tilde", body: `{"host":"127.0.0.1","port":5308,"callsign":"DL1ABC","authKey":"key","talkGroup":20,"enableLogging":true,"logFile":"~/funknetz.log"}`, detail: "Cannot write ~/funknetz.log", same: true},
+ {name: "relative", body: `{"host":"127.0.0.1","port":5308,"callsign":"DL1ABC","authKey":"key","talkGroup":20,"certificateDir":"./funknetz","enableLogging":true,"logFile":"funknetz.log"}`, detail: "Cannot write funknetz.log", same: true},
+ {name: "empty", body: `{"host":"127.0.0.1","port":5308,"callsign":"DL1ABC","authKey":"key","talkGroup":20,"certificateDir":"./funknetz","enableLogging":true,"logFile":" "}`, detail: "Cannot write logFile", same: true},
+ {name: "tilde", body: `{"host":"127.0.0.1","port":5308,"callsign":"DL1ABC","authKey":"key","talkGroup":20,"certificateDir":"./funknetz","enableLogging":true,"logFile":"~/funknetz.log"}`, detail: "Cannot write ~/funknetz.log", same: true},
{name: "enable type", body: `{"host":"127.0.0.1","port":5308,"callsign":"DL1ABC","authKey":"key","talkGroup":20,"enableLogging":"yes","logFile":"/tmp/funknetz.log"}`, detail: "enableLogging has the wrong type", same: true},
{name: "file type", body: `{"host":"127.0.0.1","port":5308,"callsign":"DL1ABC","authKey":"key","talkGroup":20,"enableLogging":true,"logFile":1}`, detail: "logFile has the wrong type", same: true},
{name: "both types", body: `{"host":"127.0.0.1","port":5308,"callsign":"DL1ABC","authKey":"key","talkGroup":20,"enableLogging":"yes","logFile":1}`, detail: "enableLogging has the wrong type", same: true},
@@ -368,6 +380,103 @@ func TestLogSettingsDoNotStopTheStation(t *testing.T) {
})
}
+func assertKeyOrder(t *testing.T, body []byte) {
+ t.Helper()
+ keys := []string{`"host"`, `"port"`, `"callsign"`, `"authKey"`, `"talkGroup"`, `"certificateDir"`, `"enableLogging"`, `"logFile"`}
+ last := -1
+ for _, key := range keys {
+ i := strings.Index(string(body), key)
+ if i < 0 || i < last {
+ t.Fatalf("key %s out of order in %s", key, body)
+ }
+ last = i
+ }
+}
+
+func TestMissingCertificateDirIsWrittenAfterTalkGroup(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,\"enableLogging\":false,\"logFile\":\"/tmp/funknetz.log\"}\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)
+ }
+ 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)
+ }
+ 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.Fatal(err)
+ }
+ if got["authKey"] != "a b" || got["callsign"] != "DL1ABC" || got["certificateDir"] != "./funknetz" {
+ t.Fatalf("fields = %#v", got)
+ }
+ if strings.Contains(string(body), t.TempDir()) {
+ t.Fatalf("file = %s, want the stored directory to stay relative", body)
+ }
+ assertKeyOrder(t, body)
+}
+
+func TestCertificateDirStopsBeforeASocket(t *testing.T) {
+ tests := []struct {
+ name string
+ dir string
+ want string
+ }{
+ {name: "wrong type", dir: "1", want: "certificateDir has the wrong type"},
+ {name: "empty", dir: `""`, want: "certificateDir is empty"},
+ {name: "whitespace", dir: `" "`, want: "certificateDir is empty"},
+ }
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ ln, err := net.Listen("tcp", "127.0.0.1:0")
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer ln.Close()
+ port := ln.Addr().(*net.TCPAddr).Port
+ body := fmt.Sprintf(`{"host":"127.0.0.1","port":%d,"callsign":"DL1ABC","authKey":"key","talkGroup":20,"certificateDir":%s,"enableLogging":false,"logFile":"/tmp/funknetz.log"}`, port, tt.dir)
+ path := filepath.Join(t.TempDir(), "funknetz.json")
+ 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)
+ }
+ t.Cleanup(st.Close)
+ got := st.State()
+ if got.Detail != tt.want || got.Connection != "Disconnected" || got.Callsign != "DL1ABC" {
+ t.Fatalf("State = %+v, want detail %q", got, tt.want)
+ }
+ after, err := os.ReadFile(path)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if string(after) != body {
+ t.Fatalf("file = %s, want it left untouched", after)
+ }
+ ctx, cancel := context.WithCancel(t.Context())
+ defer cancel()
+ go func() { _ = st.Run(ctx) }()
+ ln.(*net.TCPListener).SetDeadline(time.Now().Add(200 * time.Millisecond))
+ if conn, err := ln.Accept(); err == nil {
+ conn.Close()
+ t.Fatal("opened a socket")
+ }
+ })
+ }
+}
+
func configJSON(t testing.TB, host string, port int, callsign, authKey string, talkGroup int) string {
t.Helper()
body, err := jsonv2.Marshal(map[string]any{
diff --git a/station/lab_reflector_test.go b/station/lab_reflector_test.go
index d7f9f1a..888b638 100644
--- a/station/lab_reflector_test.go
+++ b/station/lab_reflector_test.go
@@ -2,9 +2,11 @@ package station_test
import (
"context"
+ "errors"
"net"
"os"
"strconv"
+ "syscall"
"testing"
"time"
@@ -30,6 +32,9 @@ func TestConnectsToLabReflector(t *testing.T) {
t.Skipf("lab reflector not listening on %s: %v", addr, err)
}
conn.Close()
+ if !labUDPReachable() {
+ t.Skipf("lab UDP port %s is not reachable", addr)
+ }
path := writeConfigFile(t, configJSON(t, labHost, labPort, labCallsign, labAuthKey, labTalkGroup))
st, err := station.New(station.Options{ConfigPath: path})
@@ -54,3 +59,20 @@ func TestConnectsToLabReflector(t *testing.T) {
t.Fatalf("talkGroup = %d, want %d", state.TalkGroup, labTalkGroup)
}
}
+
+func labUDPReachable() bool {
+ peer := &net.UDPAddr{IP: net.ParseIP(labHost), Port: labPort}
+ conn, err := net.DialUDP("udp", nil, peer)
+ if err != nil {
+ return false
+ }
+ defer conn.Close()
+ _, _ = conn.Write([]byte{0})
+ _ = conn.SetReadDeadline(time.Now().Add(200 * time.Millisecond))
+ _, err = conn.Read(make([]byte, 1))
+ if err == nil {
+ return true
+ }
+ var op *net.OpError
+ return !errors.As(err, &op) || !errors.Is(op.Err, syscall.ECONNREFUSED)
+}
diff --git a/station/protocol.go b/station/protocol.go
index de8c800..b93ddc7 100644
--- a/station/protocol.go
+++ b/station/protocol.go
@@ -4,26 +4,47 @@ import (
"encoding/binary"
"fmt"
"io"
+ "regexp"
+ "strconv"
)
const (
tcpHeartbeat = 1
tcpProtoVer = 5
+ tcpProtoDown = 6
tcpAuthChallenge = 10
tcpAuthResponse = 11
tcpAuthOK = 12
tcpError = 13
+ tcpStartEncReq = 14
+ tcpStartEnc = 15
+ tcpCsrReq = 16
+ tcpCsr = 17
+ tcpClientCert = 18
+ tcpCAInfo = 19
+ tcpCABundleReq = 20
+ tcpCABundle = 21
tcpServerInfo = 100
+ tcpNodeList = 101
+ tcpNodeJoined = 102
+ tcpNodeLeft = 103
tcpTalkerStart = 104
tcpTalkerStop = 105
tcpSelectTG = 106
+ tcpRequestQSY = 109
+ tcpStateEvent = 110
tcpNodeInfo = 111
+ tcpSignal = 112
+ tcpTxStatus = 113
+ tcpStartUDPEnc = 114
- udpHeartbeat = 1
- udpAudio = 101
- udpFlush = 102
+ udpHeartbeat = 1
+ udpAudio = 101
+ udpFlush = 102
+ udpAllFlushed = 103
- maxFrame = 32768
+ maxFrame = 32768
+ maxPreauth20 = 64
)
type frame struct {
@@ -45,12 +66,16 @@ func writeFrame(w io.Writer, typ uint16, body []byte) error {
}
func readFrame(r io.Reader) (frame, error) {
+ return readFrameMax(r, maxFrame)
+}
+
+func readFrameMax(r io.Reader, max uint32) (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 {
+ if n < 2 || n > max {
return frame{}, fmt.Errorf("frame length %d", n)
}
buf := make([]byte, n)
@@ -149,6 +174,48 @@ func parseServerInfo(b []byte) (uint16, []string, error) {
return id, codecs, nil
}
+func versionWire(major, minor uint16) []byte {
+ return appendU16(appendU16(nil, major), minor)
+}
+
+func parseVersion(b []byte) (uint16, uint16, error) {
+ major, rest, err := parseU16(b)
+ if err != nil {
+ return 0, 0, err
+ }
+ minor, rest, err := parseU16(rest)
+ if err != nil {
+ return 0, 0, err
+ }
+ if len(rest) != 0 {
+ return 0, 0, io.ErrUnexpectedEOF
+ }
+ return major, minor, nil
+}
+
+var atMostVersion = regexp.MustCompile(`^Unsupported protocol version \d+\.\d+\. May be at most (\d+)\.(\d+)\.$`)
+
+// fallback20 reports that the reflector's maximum is at least 2.0 and below 3.0.
+func fallback20(msg string) bool {
+ m := atMostVersion.FindStringSubmatch(msg)
+ if m == nil {
+ return false
+ }
+ major, err1 := strconv.Atoi(m[1])
+ minor, err2 := strconv.Atoi(m[2])
+ if err1 != nil || err2 != nil {
+ return false
+ }
+ return !versionBelow(major, minor, 2, 0) && versionBelow(major, minor, 3, 0)
+}
+
+func versionBelow(major, minor, maj, min int) bool {
+ if major != maj {
+ return major < maj
+ }
+ return minor < min
+}
+
func hasOPUS(codecs []string) bool {
for _, c := range codecs {
if c == "OPUS" {
diff --git a/station/protocol2_test.go b/station/protocol2_test.go
new file mode 100644
index 0000000..b62d333
--- /dev/null
+++ b/station/protocol2_test.go
@@ -0,0 +1,235 @@
+package station_test
+
+import (
+ "context"
+ "encoding/binary"
+ "testing"
+ "time"
+
+ "gitea.neitzel.de/konrad/funknetz/internal/opus"
+ "gitea.neitzel.de/konrad/funknetz/station"
+)
+
+func TestProtocol2DropsASequenceTooFarBehind(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, 32768, appendBytes(nil, pkt))
+ in.refl.writeUDP(t, 101, 7, 4, appendBytes(nil, pkt))
+ if _, err := in.st.WaitPlayback(t.Context(), opus.FrameSamples); err != nil {
+ t.Fatal(err)
+ }
+ if extra := in.st.TakePlayback(); len(extra) != 0 {
+ t.Fatalf("played %d extra samples from the old sequence", len(extra))
+ }
+}
+
+func TestProtocol2FlushCompletesOnce(t *testing.T) {
+ in := login(t, "DL1ABC", "key", 20, 7, []string{"OPUS"})
+ in.st.HoldButton()
+ in.st.ReleaseButton()
+ typ, id, _, body := in.refl.readUDP(t)
+ if typ != 102 || id != 7 || len(body) != 0 {
+ t.Fatalf("flush type %d id %d len %d", typ, id, len(body))
+ }
+ in.refl.writeUDP(t, 103, 7, 0, nil)
+ in.clock.BlockUntilWaiting(t)
+ in.clock.Advance(3 * time.Second)
+ if pkt, _, ok := tryUDP(in.refl); ok {
+ t.Fatalf("sent %d bytes after flush-complete", len(pkt))
+ }
+ if got := in.st.State(); got.Connection != "Connected" || got.Protocol != "2.0" {
+ t.Fatalf("State = %+v", got)
+ }
+}
+
+func TestProtocol2FlushTimesOutWithoutASecondFlush(t *testing.T) {
+ in := login(t, "DL1ABC", "key", 20, 7, []string{"OPUS"})
+ in.st.HoldButton()
+ in.st.ReleaseButton()
+ typ, _, _, _ := in.refl.readUDP(t)
+ if typ != 102 {
+ t.Fatalf("type %d, want flush", typ)
+ }
+ in.clock.BlockUntilWaiting(t)
+ in.clock.Advance(3 * time.Second)
+ if pkt, _, ok := tryUDP(in.refl); ok {
+ t.Fatalf("sent %d bytes when flush-complete did not arrive", len(pkt))
+ }
+}
+
+func TestSilenceFreesPushToTalkAndKeepsTheTalker(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, 0, appendBytes(nil, pkt))
+ if _, err := in.st.WaitPlayback(t.Context(), opus.FrameSamples); err != nil {
+ t.Fatal(err)
+ }
+ in.refl.writeFrame(t, 104, appendStr(appendU32(nil, 20), "DB0XYZ"))
+ if _, err := in.st.Wait(t.Context(), func(s station.State) bool { return s.Talker == "DB0XYZ" }); err != nil {
+ t.Fatal(err)
+ }
+ in.clock.BlockUntilWaiting(t)
+ in.clock.Advance(3 * time.Second)
+ in.st.HoldButton()
+ in.st.WriteMicrophone(make([]int16, opus.FrameSamples))
+ if _, _, ok := tryUDP(in.refl); ok {
+ t.Fatal("push-to-talk sent audio at exactly 3 seconds")
+ }
+ in.st.ReleaseButton()
+ in.clock.BlockUntilWaiting(t)
+ gen := in.clock.Generation()
+ in.clock.Advance(time.Second)
+ waitClock(t, in.clock, gen)
+ if got := in.st.State(); got.Talker != "DB0XYZ" || got.Transmitting {
+ t.Fatalf("State = %+v, want the talker kept and the button free", got)
+ }
+ in.st.HoldButton()
+ if !in.st.State().Transmitting {
+ t.Fatal("push-to-talk stayed blocked after the gap")
+ }
+ in.st.WriteMicrophone(make([]int16, opus.FrameSamples))
+ utyp, id, _, _ := in.refl.readUDP(t)
+ if utyp != 101 || id != 7 {
+ t.Fatalf("audio type %d id %d", utyp, id)
+ }
+}
+
+func TestUDPLimitRunsBeforeAnyDatagram(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() { _ = st.Run(ctx) }()
+ refl.accept(t)
+ declineTo2(t, refl)
+ for step := 0; step < 6; step++ {
+ clock.BlockUntilWaiting(t)
+ clock.Advance(10 * time.Second)
+ if st.State().Connection == "Disconnected" && step < 5 {
+ t.Fatalf("dropped on step %d", step)
+ }
+ replyTCP(t, refl)
+ }
+ if _, err := st.Wait(t.Context(), func(s station.State) bool { return s.Connection == "Disconnected" }); err != nil {
+ t.Fatal(err)
+ }
+ if got := st.State().Detail; got != "udp receive timeout" {
+ t.Fatalf("Detail = %q, want udp receive timeout", got)
+ }
+ clock.BlockUntilWaiting(t)
+ clock.Advance(2 * time.Second)
+ refl.accept(t)
+ typ, body := refl.readFrame(t)
+ if typ != 5 || !versionIs(body, 3, 0) {
+ t.Fatalf("retry type %d body %v, want 3.0", typ, body)
+ }
+}
+
+func TestOversizedFrameBeforeAuthRetries(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() { _ = st.Run(ctx) }()
+ refl.accept(t)
+ declineTo2(t, refl)
+ payload := make([]byte, 65)
+ var hdr [4]byte
+ binary.BigEndian.PutUint32(hdr[:], 65)
+ if _, err := refl.tcp.Write(append(hdr[:], payload...)); err != nil {
+ t.Fatal(err)
+ }
+ if _, err := st.Wait(ctx, func(s station.State) bool { return s.Connection == "Disconnected" }); err != nil {
+ t.Fatal(err)
+ }
+ clock.BlockUntilWaiting(t)
+ clock.Advance(2 * time.Second)
+ refl.accept(t)
+ typ, body := refl.readFrame(t)
+ if typ != 5 || !versionIs(body, 3, 0) {
+ t.Fatalf("retry type %d body %v, want 3.0", typ, body)
+ }
+}
+
+func TestApplicationMessageBeforeLoginRetries(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() { _ = st.Run(ctx) }()
+ refl.accept(t)
+ declineTo2(t, refl)
+ refl.writeFrame(t, 100, serverInfo(1, nil, []string{"OPUS"}))
+ got, err := st.Wait(t.Context(), func(s station.State) bool { return s.Detail == "Protocol error" })
+ if err != nil {
+ t.Fatal(err)
+ }
+ if got.Connection != "Disconnected" || got.Protocol != "" {
+ t.Fatalf("State = %+v", got)
+ }
+ clock.BlockUntilWaiting(t)
+ clock.Advance(2 * time.Second)
+ refl.accept(t)
+ typ, body := refl.readFrame(t)
+ if typ != 5 || !versionIs(body, 3, 0) {
+ t.Fatalf("retry type %d body %v", typ, body)
+ }
+ if st.State().Detail != "Protocol error" {
+ t.Fatalf("Detail = %q, want the reason kept", st.State().Detail)
+ }
+}
+
+func TestIgnoredNoticesDoNotChangeTheStation(t *testing.T) {
+ in := login(t, "DL1ABC", "key", 20, 7, []string{"OPUS"})
+ in.refl.writeFrame(t, 101, appendStr(appendU16(nil, 1), "DB0XYZ"))
+ in.refl.writeFrame(t, 102, appendStr(nil, "DB0XYZ"))
+ in.refl.writeFrame(t, 103, appendStr(nil, "DB0XYZ"))
+ in.refl.writeFrame(t, 109, appendU32(nil, 99))
+ body := appendStr(nil, "src")
+ body = appendStr(body, "name")
+ body = appendStr(body, "msg")
+ in.refl.writeFrame(t, 110, body)
+ in.refl.writeFrame(t, 112, append(appendU16(nil, 1), []byte{1, 40, 1}...))
+ in.refl.writeFrame(t, 113, append(appendU16(nil, 1), []byte{1, 1}...))
+ in.clock.BlockUntilWaiting(t)
+ in.clock.Advance(10 * time.Second)
+ typ, _ := in.refl.readFrame(t)
+ if typ != 1 {
+ t.Fatalf("TCP type %d, want a heartbeat", typ)
+ }
+ got := in.st.State()
+ if got.Connection != "Connected" || got.Protocol != "2.0" || got.TalkGroup != 20 || got.Talker != "" {
+ t.Fatalf("State = %+v", got)
+ }
+}
diff --git a/station/protocol3_test.go b/station/protocol3_test.go
new file mode 100644
index 0000000..4e962a2
--- /dev/null
+++ b/station/protocol3_test.go
@@ -0,0 +1,1000 @@
+package station_test
+
+import (
+ "bytes"
+ "context"
+ "crypto"
+ "crypto/aes"
+ "crypto/cipher"
+ "crypto/hmac"
+ "crypto/rand"
+ "crypto/rsa"
+ "crypto/sha1"
+ "crypto/sha256"
+ "crypto/tls"
+ "crypto/x509"
+ "crypto/x509/pkix"
+ "encoding/asn1"
+ "encoding/binary"
+ "encoding/json"
+ "encoding/pem"
+ "errors"
+ "math/big"
+ "net"
+ "os"
+ "path/filepath"
+ "testing"
+ "time"
+
+ jsonv2 "encoding/json/v2"
+ "golang.org/x/sys/unix"
+
+ "gitea.neitzel.de/konrad/funknetz/internal/opus"
+ "gitea.neitzel.de/konrad/funknetz/station"
+)
+
+func configWithDir(t testing.TB, host string, port int, callsign, authKey string, talkGroup int, certDir, logPath string) string {
+ t.Helper()
+ if logPath == "" {
+ logPath = "/tmp/funknetz-test.log"
+ }
+ body, err := jsonv2.Marshal(map[string]any{
+ "host": host,
+ "port": port,
+ "callsign": callsign,
+ "authKey": authKey,
+ "talkGroup": talkGroup,
+ "certificateDir": certDir,
+ "enableLogging": false,
+ "logFile": logPath,
+ })
+ if err != nil {
+ t.Fatal(err)
+ }
+ return string(body)
+}
+
+type testPKI struct {
+ caPEM []byte
+ caCert *x509.Certificate
+ caKey *rsa.PrivateKey
+ server tls.Certificate
+ sig []byte
+}
+
+func newPKI(t *testing.T, dns string, ip net.IP) *testPKI {
+ t.Helper()
+ caKey, err := rsa.GenerateKey(rand.Reader, 2048)
+ if err != nil {
+ t.Fatal(err)
+ }
+ caTmpl := &x509.Certificate{
+ SerialNumber: big.NewInt(1),
+ Subject: pkix.Name{CommonName: "Test CA"},
+ IsCA: true,
+ BasicConstraintsValid: true,
+ NotBefore: time.Now().Add(-time.Hour),
+ NotAfter: time.Now().Add(24 * time.Hour),
+ KeyUsage: x509.KeyUsageCertSign | x509.KeyUsageDigitalSignature,
+ }
+ caDER, err := x509.CreateCertificate(rand.Reader, caTmpl, caTmpl, &caKey.PublicKey, caKey)
+ if err != nil {
+ t.Fatal(err)
+ }
+ caCert, err := x509.ParseCertificate(caDER)
+ if err != nil {
+ t.Fatal(err)
+ }
+ caPEM := pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: caDER})
+ srvKey, err := rsa.GenerateKey(rand.Reader, 2048)
+ if err != nil {
+ t.Fatal(err)
+ }
+ srvTmpl := &x509.Certificate{
+ SerialNumber: big.NewInt(2),
+ Subject: pkix.Name{CommonName: dns},
+ NotBefore: time.Now().Add(-time.Hour),
+ NotAfter: time.Now().Add(24 * time.Hour),
+ KeyUsage: x509.KeyUsageDigitalSignature | x509.KeyUsageKeyEncipherment,
+ ExtKeyUsage: []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth},
+ }
+ if dns != "" {
+ srvTmpl.DNSNames = []string{dns}
+ }
+ if ip != nil {
+ srvTmpl.IPAddresses = []net.IP{ip}
+ }
+ srvDER, err := x509.CreateCertificate(rand.Reader, srvTmpl, caCert, &srvKey.PublicKey, caKey)
+ if err != nil {
+ t.Fatal(err)
+ }
+ srvPEM := pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: srvDER})
+ keyPEM := pem.EncodeToMemory(&pem.Block{Type: "RSA PRIVATE KEY", Bytes: x509.MarshalPKCS1PrivateKey(srvKey)})
+ cert, err := tls.X509KeyPair(srvPEM, keyPEM)
+ if err != nil {
+ t.Fatal(err)
+ }
+ sum := sha256.Sum256(caPEM)
+ sig, err := rsa.SignPKCS1v15(rand.Reader, caKey, crypto.SHA256, sum[:])
+ if err != nil {
+ t.Fatal(err)
+ }
+ return &testPKI{caPEM: caPEM, caCert: caCert, caKey: caKey, server: cert, sig: sig}
+}
+
+func (p *testPKI) info() []byte {
+ sum := sha256.Sum256(p.caPEM)
+ b := appendU16(nil, uint16(len(p.caPEM)))
+ return appendBytes(b, sum[:])
+}
+
+func (p *testPKI) bundle(sig []byte) []byte {
+ b := appendStr(nil, string(p.caPEM))
+ b = appendBytes(b, sig)
+ return appendStr(b, string(p.caPEM))
+}
+
+func (p *testPKI) serverConfig() *tls.Config {
+ return &tls.Config{
+ Certificates: []tls.Certificate{p.server},
+ MinVersion: tls.VersionTLS12,
+ ClientAuth: tls.RequestClientCert,
+ }
+}
+
+func signClient(t *testing.T, p *testPKI, key *rsa.PrivateKey, cn string) []byte {
+ t.Helper()
+ tmpl := &x509.Certificate{
+ SerialNumber: big.NewInt(3),
+ Subject: pkix.Name{CommonName: cn},
+ NotBefore: time.Now().Add(-time.Hour),
+ NotAfter: time.Now().Add(24 * time.Hour),
+ KeyUsage: x509.KeyUsageDigitalSignature | x509.KeyUsageKeyEncipherment | x509.KeyUsageKeyAgreement,
+ ExtKeyUsage: []x509.ExtKeyUsage{x509.ExtKeyUsageClientAuth},
+ }
+ der, err := x509.CreateCertificate(rand.Reader, tmpl, p.caCert, &key.PublicKey, p.caKey)
+ if err != nil {
+ t.Fatal(err)
+ }
+ return pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: der})
+}
+
+func writeKey(t *testing.T, path string, key *rsa.PrivateKey) {
+ t.Helper()
+ body := pem.EncodeToMemory(&pem.Block{Type: "RSA PRIVATE KEY", Bytes: x509.MarshalPKCS1PrivateKey(key)})
+ if err := os.WriteFile(path, body, 0o600); err != nil {
+ t.Fatal(err)
+ }
+}
+
+type v3 struct {
+ st *station.Station
+ clock *fakeClock
+ refl *reflector
+ pki *testPKI
+ certDir string
+ cancel context.CancelFunc
+ id uint16
+ iv []byte
+ key []byte
+ reg []byte
+}
+
+func startV3(t *testing.T, certDir string) *v3 {
+ t.Helper()
+ refl := newReflector(t)
+ if certDir == "" {
+ certDir = filepath.Join(t.TempDir(), "certs")
+ }
+ path := writeConfigFile(t, configWithDir(t, "127.0.0.1", refl.port, "DL1ABC", "key", 20, certDir, ""))
+ 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) }()
+ pki := newPKI(t, "", net.ParseIP("127.0.0.1"))
+ return &v3{st: st, clock: clock, refl: refl, pki: pki, certDir: certDir, cancel: cancel}
+}
+
+func (v *v3) throughBundle(t *testing.T) {
+ t.Helper()
+ v.refl.accept(t)
+ typ, body := v.refl.readFrame(t)
+ if typ != 5 || !versionIs(body, 3, 0) {
+ t.Fatalf("version type %d body %v", typ, body)
+ }
+ v.refl.writeFrame(t, 19, v.pki.info())
+ typ, _ = v.refl.readFrame(t)
+ if typ != 20 {
+ t.Fatalf("type %d, want a CA bundle request", typ)
+ }
+ info, err := os.Stat(v.certDir)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if info.Mode().Perm() != 0o700 {
+ t.Fatalf("directory mode %o, want 0700", info.Mode().Perm())
+ }
+ v.refl.writeFrame(t, 21, v.pki.bundle(v.pki.sig))
+ typ, _ = v.refl.readFrame(t)
+ if typ != 14 {
+ t.Fatalf("type %d, want start encryption", typ)
+ }
+}
+
+func (v *v3) tls(t *testing.T, cfg *tls.Config) {
+ t.Helper()
+ v.refl.writeFrame(t, 15, nil)
+ if err := v.refl.handshake(t, cfg); err != nil {
+ t.Fatal(err)
+ }
+}
+
+func (v *v3) hmac(t *testing.T) {
+ t.Helper()
+ v.refl.writeFrame(t, 16, nil)
+ typ, body := v.refl.readFrame(t)
+ if typ != 17 {
+ t.Fatalf("type %d, want the signing request", typ)
+ }
+ checkCSR(t, body, v.certDir)
+ challenge := make([]byte, 20)
+ for i := range challenge {
+ challenge[i] = byte(i + 1)
+ }
+ v.refl.writeFrame(t, 10, appendBytes(nil, challenge))
+ typ, body = v.refl.readFrame(t)
+ if typ != 11 {
+ t.Fatalf("type %d, want auth response", typ)
+ }
+ call, rest := takeStr(t, body)
+ digest, _ := takeBytes(t, rest)
+ if call != "DL1ABC" {
+ t.Fatalf("callsign %q", call)
+ }
+ mac := hmacSum([]byte("key"), challenge)
+ if !bytes.Equal(digest, mac) {
+ t.Fatal("auth digest does not match the auth key")
+ }
+ v.refl.writeFrame(t, 12, nil)
+}
+
+func (v *v3) serverAndKey(t *testing.T, codecs []string) {
+ t.Helper()
+ v.refl.writeFrame(t, 100, serverInfo(0x1234, []string{"db0abc"}, codecs))
+ typ, body := v.refl.readFrame(t)
+ if typ != 111 {
+ t.Fatalf("type %d, want node information", typ)
+ }
+ iv, rest := takeBytes(t, body)
+ key, rest := takeBytes(t, rest)
+ raw, rest := takeStr(t, rest)
+ if len(iv) != 6 || len(key) != 16 || len(rest) != 0 {
+ t.Fatalf("iv %d key %d rest %d", len(iv), len(key), len(rest))
+ }
+ var got map[string]string
+ if err := json.Unmarshal([]byte(raw), &got); err != nil {
+ t.Fatal(err)
+ }
+ if got["sw"] != station.SoftwareName || got["swVer"] != station.SoftwareVersion || got["projVer"] != station.SoftwareVersion {
+ t.Fatalf("node json %#v", got)
+ }
+ var u unix.Utsname
+ if err := unix.Uname(&u); err != nil {
+ t.Fatal(err)
+ }
+ name := u.Machine[:]
+ if i := bytes.IndexByte(name, 0); i >= 0 {
+ name = name[:i]
+ }
+ if got["machineArch"] != string(name) {
+ t.Fatalf("machineArch %q, want %q", got["machineArch"], name)
+ }
+ v.iv = iv
+ v.key = key
+ v.id = 0x1234
+ if _, _, ok := tryFrame(v.refl); ok {
+ t.Fatal("selected the talk group before the UDP heartbeat")
+ }
+ v.refl.writeFrame(t, 114, nil)
+ reg := v.refl.readRawUDP(t)
+ plain := openReg(t, v.key, v.iv, v.id, reg)
+ if binary.BigEndian.Uint16(plain) != 1 {
+ t.Fatalf("registration type %d", binary.BigEndian.Uint16(plain))
+ }
+ v.reg = reg
+ if got := v.st.State(); got.Connection == "Connected" || got.Protocol != "" {
+ t.Fatalf("State = %+v, want not connected before the heartbeat", got)
+ }
+}
+
+func (v *v3) beat(t *testing.T) {
+ t.Helper()
+ v.refl.writeRawUDP(t, sealToStation(t, v.key, v.iv, 0, appendU16(nil, 1)))
+ typ, body := v.refl.readFrame(t)
+ if typ != 106 {
+ t.Fatalf("type %d, want select talk group", typ)
+ }
+ tg, _ := takeU32(t, body)
+ if tg != 20 {
+ t.Fatalf("talk group %d", tg)
+ }
+ got, err := v.st.Wait(t.Context(), func(s station.State) bool { return s.Connection == "Connected" })
+ if err != nil {
+ t.Fatal(err)
+ }
+ if got.Protocol != "3.0" || got.Callsign != "DL1ABC" || got.TalkGroup != 20 {
+ t.Fatalf("State = %+v", got)
+ }
+}
+
+func checkCSR(t *testing.T, body []byte, dir string) {
+ t.Helper()
+ raw, rest := takeStr(t, body)
+ if len(rest) != 0 {
+ t.Fatalf("%d trailing bytes", len(rest))
+ }
+ block, _ := pem.Decode([]byte(raw))
+ if block == nil || block.Type != "CERTIFICATE REQUEST" {
+ t.Fatal("signing request is not a PEM request")
+ }
+ req, err := x509.ParseCertificateRequest(block.Bytes)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if err := req.CheckSignature(); err != nil {
+ t.Fatal(err)
+ }
+ if req.Subject.CommonName != "DL1ABC" || len(req.EmailAddresses) != 0 || len(req.DNSNames) != 0 {
+ t.Fatalf("subject %+v emails %v", req.Subject, req.EmailAddresses)
+ }
+ var sawBC, sawKU, sawEKU bool
+ for _, ext := range req.Extensions {
+ switch {
+ case ext.Id.Equal(asn1.ObjectIdentifier{2, 5, 29, 19}):
+ sawBC = ext.Critical
+ case ext.Id.Equal(asn1.ObjectIdentifier{2, 5, 29, 15}):
+ sawKU = ext.Critical
+ var bits asn1.BitString
+ if _, err := asn1.Unmarshal(ext.Value, &bits); err != nil {
+ t.Fatal(err)
+ }
+ if bits.BitLength < 5 || bits.Bytes[0]&0xA8 != 0xA8 {
+ t.Fatalf("key usage %x", bits.Bytes)
+ }
+ case ext.Id.Equal(asn1.ObjectIdentifier{2, 5, 29, 37}):
+ sawEKU = true
+ }
+ }
+ if !sawBC || !sawKU || !sawEKU {
+ t.Fatalf("extensions basic %v key %v eku %v", sawBC, sawKU, sawEKU)
+ }
+ info, err := os.Stat(filepath.Join(dir, "key.pem"))
+ if err != nil {
+ t.Fatal(err)
+ }
+ if info.Mode().Perm() != 0o600 {
+ t.Fatalf("key mode %o, want 0600", info.Mode().Perm())
+ }
+ keyPEM, err := os.ReadFile(filepath.Join(dir, "key.pem"))
+ if err != nil {
+ t.Fatal(err)
+ }
+ kblock, _ := pem.Decode(keyPEM)
+ key, err := x509.ParsePKCS1PrivateKey(kblock.Bytes)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if key.N.BitLen() != 2048 {
+ t.Fatalf("key bits %d", key.N.BitLen())
+ }
+}
+
+func openReg(t *testing.T, key, iv []byte, id uint16, pkt []byte) []byte {
+ t.Helper()
+ if len(pkt) < 14 {
+ t.Fatalf("datagram %d", len(pkt))
+ }
+ return openGCM(t, key, iv, id, 0, pkt[:6], pkt[6:14], pkt[14:])
+}
+
+func openGCM(t *testing.T, key, iv []byte, id uint16, counter uint32, aad, tag, ct []byte) []byte {
+ t.Helper()
+ plain, err := openTruncTest(key, nonceFor(iv, id, counter), tag, ct, aad)
+ if err != nil {
+ t.Fatal(err)
+ }
+ return plain
+}
+
+func sealToStation(t *testing.T, key, iv []byte, counter uint32, plain []byte) []byte {
+ t.Helper()
+ var aad [4]byte
+ binary.BigEndian.PutUint32(aad[:], counter)
+ tag, ct, err := sealTruncTest(key, nonceFor(iv, 0, counter), plain, aad[:])
+ if err != nil {
+ t.Fatal(err)
+ }
+ wire := append([]byte{}, aad[:]...)
+ wire = append(wire, tag...)
+ return append(wire, ct...)
+}
+
+func nonceFor(iv []byte, id uint16, counter uint32) []byte {
+ nonce := make([]byte, 12)
+ copy(nonce, iv)
+ binary.BigEndian.PutUint16(nonce[6:8], id)
+ binary.BigEndian.PutUint32(nonce[8:12], counter)
+ return nonce
+}
+
+func sealTruncTest(key, nonce, plain, aad []byte) (tag, ct []byte, err error) {
+ block, err := aes.NewCipher(key)
+ if err != nil {
+ return nil, nil, err
+ }
+ gcm, err := cipher.NewGCM(block)
+ if err != nil {
+ return nil, nil, err
+ }
+ out := gcm.Seal(nil, nonce, plain, aad)
+ return append([]byte(nil), out[len(plain):len(plain)+8]...), append([]byte(nil), out[:len(plain)]...), nil
+}
+
+func openTruncTest(key, nonce, tag, ct, aad []byte) ([]byte, error) {
+ block, err := aes.NewCipher(key)
+ if err != nil {
+ return nil, err
+ }
+ gcm, err := cipher.NewGCM(block)
+ if err != nil {
+ return nil, err
+ }
+ stream := gcm.Seal(nil, nonce, make([]byte, len(ct)), aad)
+ plain := make([]byte, len(ct))
+ for i := range plain {
+ plain[i] = ct[i] ^ stream[i]
+ }
+ check := gcm.Seal(nil, nonce, plain, aad)
+ if !bytes.Equal(check[len(plain):len(plain)+8], tag) {
+ return nil, errors.New("tag")
+ }
+ return plain, nil
+}
+
+func openFromStation(t *testing.T, key, iv []byte, id uint16, pkt []byte) (uint32, []byte) {
+ t.Helper()
+ if len(pkt) < 12 {
+ t.Fatalf("datagram %d", len(pkt))
+ }
+ counter := binary.BigEndian.Uint32(pkt[:4])
+ return counter, openGCM(t, key, iv, id, counter, pkt[:4], pkt[4:12], pkt[12:])
+}
+
+func TestCABundleIsStoredWhenTheSignatureChecks(t *testing.T) {
+ v := startV3(t, "")
+ v.throughBundle(t)
+ got, err := os.ReadFile(filepath.Join(v.certDir, "ca.pem"))
+ if err != nil {
+ t.Fatal(err)
+ }
+ if !bytes.Equal(got, v.pki.caPEM) {
+ t.Fatal("stored bundle does not match the signed PEM")
+ }
+}
+
+func TestBadCASignatureRetriesOn3(t *testing.T) {
+ v := startV3(t, "")
+ v.refl.accept(t)
+ v.refl.readFrame(t)
+ v.refl.writeFrame(t, 19, v.pki.info())
+ typ, _ := v.refl.readFrame(t)
+ if typ != 20 {
+ t.Fatalf("type %d, want a bundle request", typ)
+ }
+ sig := append([]byte{}, v.pki.sig...)
+ sig[0] ^= 0xff
+ v.refl.writeFrame(t, 21, v.pki.bundle(sig))
+ got, err := v.st.Wait(context.Background(), func(s station.State) bool { return s.Detail == "CA signature rejected" })
+ if err != nil {
+ t.Fatal(err)
+ }
+ if got.Connection != "Disconnected" {
+ t.Fatalf("Connection %s", got.Connection)
+ }
+ if _, err := os.Stat(filepath.Join(v.certDir, "ca.pem")); !os.IsNotExist(err) {
+ t.Fatalf("ca.pem err = %v, want absent", err)
+ }
+ v.clock.BlockUntilWaiting(t)
+ v.refl.ln.(*net.TCPListener).SetDeadline(time.Now().Add(150 * time.Millisecond))
+ if conn, err := v.refl.ln.Accept(); err == nil {
+ conn.Close()
+ t.Fatal("retried before 2 seconds")
+ }
+ v.clock.Advance(2 * time.Second)
+ v.refl.accept(t)
+ typ, body := v.refl.readFrame(t)
+ if typ != 5 || !versionIs(body, 3, 0) {
+ t.Fatalf("retry type %d body %v", typ, body)
+ }
+ if v.st.State().Detail != "CA signature rejected" {
+ t.Fatalf("Detail = %q", v.st.State().Detail)
+ }
+}
+
+func TestExistingCABundleIsKept(t *testing.T) {
+ v := startV3(t, "")
+ if err := os.Mkdir(v.certDir, 0o700); err != nil {
+ t.Fatal(err)
+ }
+ if err := os.WriteFile(filepath.Join(v.certDir, "ca.pem"), v.pki.caPEM, 0o644); err != nil {
+ t.Fatal(err)
+ }
+ v.refl.accept(t)
+ v.refl.readFrame(t)
+ v.refl.writeFrame(t, 19, v.pki.info())
+ typ, _ := v.refl.readFrame(t)
+ if typ != 14 {
+ t.Fatalf("type %d, want start encryption and no bundle request", typ)
+ }
+}
+
+func TestCADigestMismatchIsLoggedAndKept(t *testing.T) {
+ logPath := filepath.Join(t.TempDir(), "funknetz.log")
+ refl := newReflector(t)
+ certDir := filepath.Join(t.TempDir(), "certs")
+ if err := os.Mkdir(certDir, 0o700); err != nil {
+ t.Fatal(err)
+ }
+ old := []byte("not the announced bundle\n")
+ if err := os.WriteFile(filepath.Join(certDir, "ca.pem"), old, 0o644); err != nil {
+ t.Fatal(err)
+ }
+ body, err := jsonv2.Marshal(map[string]any{
+ "host": "127.0.0.1", "port": refl.port, "callsign": "DL1ABC", "authKey": "key",
+ "talkGroup": 20, "certificateDir": certDir, "enableLogging": true, "logFile": logPath,
+ })
+ if err != nil {
+ t.Fatal(err)
+ }
+ path := writeConfigFile(t, string(body))
+ 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) }()
+ pki := newPKI(t, "", net.ParseIP("127.0.0.1"))
+ refl.accept(t)
+ refl.readFrame(t)
+ refl.writeFrame(t, 19, pki.info())
+ typ, _ := refl.readFrame(t)
+ if typ != 14 {
+ t.Fatalf("type %d, want start encryption", typ)
+ }
+ got, err := os.ReadFile(filepath.Join(certDir, "ca.pem"))
+ if err != nil {
+ t.Fatal(err)
+ }
+ if !bytes.Equal(got, old) {
+ t.Fatal("replaced the stored bundle")
+ }
+ logBody, err := os.ReadFile(logPath)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if !bytes.Contains(logBody, []byte("CA digest differs")) {
+ t.Fatalf("log = %q", logBody)
+ }
+}
+
+func TestCannotCreateCertificateDirStops(t *testing.T) {
+ refl := newReflector(t)
+ certDir := filepath.Join(t.TempDir(), "blocked")
+ if err := os.WriteFile(certDir, []byte("x"), 0o600); err != nil {
+ t.Fatal(err)
+ }
+ path := writeConfigFile(t, configWithDir(t, "127.0.0.1", refl.port, "DL1ABC", "key", 20, certDir, ""))
+ 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) }()
+ pki := newPKI(t, "", net.ParseIP("127.0.0.1"))
+ refl.accept(t)
+ refl.readFrame(t)
+ refl.writeFrame(t, 19, pki.info())
+ got, err := st.Wait(ctx, func(s station.State) bool {
+ return len(s.Detail) >= len("Cannot create ") && s.Detail[:len("Cannot create ")] == "Cannot create "
+ })
+ if err != nil {
+ t.Fatal(err)
+ }
+ if got.Detail != "Cannot create "+certDir || got.Connection != "Disconnected" {
+ t.Fatalf("State = %+v", got)
+ }
+ 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("opened a 2.0 session")
+ }
+}
+
+func TestRelativeCertificateDirStaysRelative(t *testing.T) {
+ refl := newReflector(t)
+ dir := t.TempDir()
+ path := filepath.Join(dir, "funknetz.json")
+ body := configWithDir(t, "127.0.0.1", refl.port, "DL1ABC", "key", 20, "./funknetz", "")
+ if err := os.WriteFile(path, []byte(body), 0o600); err != nil {
+ t.Fatal(err)
+ }
+ before := body
+ 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) }()
+ pki := newPKI(t, "", net.ParseIP("127.0.0.1"))
+ refl.accept(t)
+ refl.readFrame(t)
+ refl.writeFrame(t, 19, pki.info())
+ typ, _ := refl.readFrame(t)
+ if typ != 20 {
+ t.Fatalf("type %d, want a bundle request", typ)
+ }
+ info, err := os.Stat(filepath.Join(dir, "funknetz"))
+ if err != nil {
+ t.Fatal(err)
+ }
+ if !info.IsDir() || info.Mode().Perm() != 0o700 {
+ t.Fatalf("mode %o", info.Mode().Perm())
+ }
+ after, err := os.ReadFile(path)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if string(after) != before {
+ t.Fatalf("file = %s, want the relative path kept", after)
+ }
+ cwd, err := os.Getwd()
+ if err != nil {
+ t.Fatal(err)
+ }
+ if _, err := os.Stat(filepath.Join(cwd, "funknetz")); err == nil {
+ t.Fatal("created funknetz in the working directory")
+ }
+}
+
+func TestTLSUsesTheCallsignCertificate(t *testing.T) {
+ v := startV3(t, "")
+ if err := os.Mkdir(v.certDir, 0o700); err != nil {
+ t.Fatal(err)
+ }
+ key, err := rsa.GenerateKey(rand.Reader, 2048)
+ if err != nil {
+ t.Fatal(err)
+ }
+ writeKey(t, filepath.Join(v.certDir, "key.pem"), key)
+ if err := os.WriteFile(filepath.Join(v.certDir, "cert.pem"), signClient(t, v.pki, key, "DL1ABC"), 0o644); err != nil {
+ t.Fatal(err)
+ }
+ v.throughBundle(t)
+ v.tls(t, v.pki.serverConfig())
+ peer := v.refl.tls.ConnectionState().PeerCertificates
+ if len(peer) != 1 || peer[0].Subject.CommonName != "DL1ABC" {
+ t.Fatalf("presented %#v", peer)
+ }
+ v.refl.writeFrame(t, 12, nil)
+ v.serverAndKey(t, []string{"OPUS"})
+ v.beat(t)
+}
+
+func TestCertificateProblemsStop(t *testing.T) {
+ cases := []struct {
+ name string
+ prep func(t *testing.T, dir string, p *testPKI)
+ want string
+ }{
+ {name: "callsign", want: "Certificate callsign does not match", prep: func(t *testing.T, dir string, p *testPKI) {
+ key, err := rsa.GenerateKey(rand.Reader, 2048)
+ if err != nil {
+ t.Fatal(err)
+ }
+ writeKey(t, filepath.Join(dir, "key.pem"), key)
+ if err := os.WriteFile(filepath.Join(dir, "cert.pem"), signClient(t, p, key, "DB0ZZZ"), 0o644); err != nil {
+ t.Fatal(err)
+ }
+ }},
+ {name: "missing key", want: "Certificate key is missing", prep: func(t *testing.T, dir string, p *testPKI) {
+ key, err := rsa.GenerateKey(rand.Reader, 2048)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if err := os.WriteFile(filepath.Join(dir, "cert.pem"), signClient(t, p, key, "DL1ABC"), 0o644); err != nil {
+ t.Fatal(err)
+ }
+ }},
+ {name: "mismatch", want: "Certificate does not match the key", prep: func(t *testing.T, dir string, p *testPKI) {
+ key, err := rsa.GenerateKey(rand.Reader, 2048)
+ if err != nil {
+ t.Fatal(err)
+ }
+ other, err := rsa.GenerateKey(rand.Reader, 2048)
+ if err != nil {
+ t.Fatal(err)
+ }
+ writeKey(t, filepath.Join(dir, "key.pem"), other)
+ if err := os.WriteFile(filepath.Join(dir, "cert.pem"), signClient(t, p, key, "DL1ABC"), 0o644); err != nil {
+ t.Fatal(err)
+ }
+ }},
+ }
+ for _, tc := range cases {
+ t.Run(tc.name, func(t *testing.T) {
+ v := startV3(t, "")
+ if err := os.Mkdir(v.certDir, 0o700); err != nil {
+ t.Fatal(err)
+ }
+ tc.prep(t, v.certDir, v.pki)
+ v.throughBundle(t)
+ v.refl.writeFrame(t, 15, nil)
+ got, err := v.st.Wait(context.Background(), func(s station.State) bool { return s.Detail == tc.want })
+ if err != nil {
+ t.Fatal(err)
+ }
+ if got.Connection != "Disconnected" {
+ t.Fatalf("Connection %s", got.Connection)
+ }
+ v.clock.Advance(2 * time.Second)
+ v.refl.ln.(*net.TCPListener).SetDeadline(time.Now().Add(150 * time.Millisecond))
+ if conn, err := v.refl.ln.Accept(); err == nil {
+ conn.Close()
+ t.Fatal("retried")
+ }
+ })
+ }
+}
+
+func TestHostMismatchRetriesOn3(t *testing.T) {
+ v := startV3(t, "")
+ v.pki = newPKI(t, "example.com", nil)
+ v.throughBundle(t)
+ v.refl.writeFrame(t, 15, nil)
+ _ = v.refl.handshake(t, v.pki.serverConfig())
+ got, err := v.st.Wait(context.Background(), func(s station.State) bool {
+ return s.Detail == "Reflector certificate does not match the host"
+ })
+ if err != nil {
+ t.Fatal(err)
+ }
+ if got.Connection != "Disconnected" {
+ t.Fatalf("Connection %s", got.Connection)
+ }
+ v.clock.BlockUntilWaiting(t)
+ v.clock.Advance(2 * time.Second)
+ v.refl.accept(t)
+ typ, body := v.refl.readFrame(t)
+ if typ != 5 || !versionIs(body, 3, 0) {
+ t.Fatalf("retry type %d body %v", typ, body)
+ }
+ if v.st.State().Detail != "Reflector certificate does not match the host" {
+ t.Fatalf("Detail = %q", v.st.State().Detail)
+ }
+}
+
+func TestTLSHandshakeFailureRetriesOn3(t *testing.T) {
+ v := startV3(t, "")
+ v.throughBundle(t)
+ cfg := v.pki.serverConfig()
+ cfg.MinVersion = tls.VersionTLS10
+ cfg.MaxVersion = tls.VersionTLS11
+ v.refl.writeFrame(t, 15, nil)
+ _ = v.refl.handshake(t, cfg)
+ if _, err := v.st.Wait(context.Background(), func(s station.State) bool { return s.Detail == "TLS handshake failed" }); err != nil {
+ t.Fatal(err)
+ }
+ v.clock.BlockUntilWaiting(t)
+ v.clock.Advance(2 * time.Second)
+ v.refl.accept(t)
+ typ, body := v.refl.readFrame(t)
+ if typ != 5 || !versionIs(body, 3, 0) {
+ t.Fatalf("retry type %d body %v", typ, body)
+ }
+}
+
+func TestSigningRequestAndAuthKey(t *testing.T) {
+ v := startV3(t, "")
+ v.throughBundle(t)
+ v.tls(t, v.pki.serverConfig())
+ v.hmac(t)
+ v.serverAndKey(t, []string{"OPUS"})
+ v.beat(t)
+}
+
+func TestReflectorCertificateIsSavedAndReconnects(t *testing.T) {
+ v := startV3(t, "")
+ v.throughBundle(t)
+ v.tls(t, v.pki.serverConfig())
+ v.refl.writeFrame(t, 16, nil)
+ typ, _ := v.refl.readFrame(t)
+ if typ != 17 {
+ t.Fatalf("type %d, want the signing request", typ)
+ }
+ pemCert := signClient(t, v.pki, mustKey(t), "DL1ABC")
+ v.refl.writeFrame(t, 18, appendStr(nil, string(pemCert)))
+ v.refl.accept(t)
+ typ, body := v.refl.readFrame(t)
+ if typ != 5 || !versionIs(body, 3, 0) {
+ t.Fatalf("reconnect type %d body %v", typ, body)
+ }
+ got, err := os.ReadFile(filepath.Join(v.certDir, "cert.pem"))
+ if err != nil {
+ t.Fatal(err)
+ }
+ if !bytes.Equal(got, pemCert) {
+ t.Fatal("certificate file does not match the reflector PEM")
+ }
+}
+
+func TestCertificateAfterLoginReconnects(t *testing.T) {
+ v := startV3(t, "")
+ v.throughBundle(t)
+ v.tls(t, v.pki.serverConfig())
+ v.hmac(t)
+ v.serverAndKey(t, []string{"OPUS"})
+ v.beat(t)
+ pemCert := signClient(t, v.pki, mustKey(t), "DL1ABC")
+ v.refl.writeFrame(t, 18, appendStr(nil, string(pemCert)))
+ v.refl.accept(t)
+ typ, body := v.refl.readFrame(t)
+ if typ != 5 || !versionIs(body, 3, 0) {
+ t.Fatalf("reconnect type %d body %v", typ, body)
+ }
+}
+
+func TestNewerSigningRequestIsSentAfterLogin(t *testing.T) {
+ v := startV3(t, "")
+ if err := os.Mkdir(v.certDir, 0o700); err != nil {
+ t.Fatal(err)
+ }
+ key, err := rsa.GenerateKey(rand.Reader, 2048)
+ if err != nil {
+ t.Fatal(err)
+ }
+ writeKey(t, filepath.Join(v.certDir, "key.pem"), key)
+ if err := os.WriteFile(filepath.Join(v.certDir, "cert.pem"), signClient(t, v.pki, key, "DL1ABC"), 0o644); err != nil {
+ t.Fatal(err)
+ }
+ csr := []byte("-----BEGIN CERTIFICATE REQUEST-----\nMIIB\n-----END CERTIFICATE REQUEST-----\n")
+ if err := os.WriteFile(filepath.Join(v.certDir, "csr.pem"), csr, 0o644); err != nil {
+ t.Fatal(err)
+ }
+ future := time.Now().Add(time.Hour)
+ if err := os.Chtimes(filepath.Join(v.certDir, "csr.pem"), future, future); err != nil {
+ t.Fatal(err)
+ }
+ v.throughBundle(t)
+ v.tls(t, v.pki.serverConfig())
+ v.refl.writeFrame(t, 12, nil)
+ typ, body := v.refl.readFrame(t)
+ if typ != 17 {
+ t.Fatalf("type %d, want the newer signing request", typ)
+ }
+ raw, _ := takeStr(t, body)
+ if raw != string(csr) {
+ t.Fatalf("csr = %q", raw)
+ }
+}
+
+func TestMissingOpusOn3Stops(t *testing.T) {
+ v := startV3(t, "")
+ v.throughBundle(t)
+ v.tls(t, v.pki.serverConfig())
+ v.hmac(t)
+ v.refl.writeFrame(t, 100, serverInfo(1, nil, []string{"GSM"}))
+ if _, err := v.st.Wait(context.Background(), func(s station.State) bool {
+ return s.Detail == "The reflector did not offer OPUS."
+ }); err != nil {
+ t.Fatal(err)
+ }
+ v.clock.Advance(2 * time.Second)
+ v.refl.ln.(*net.TCPListener).SetDeadline(time.Now().Add(150 * time.Millisecond))
+ if conn, err := v.refl.ln.Accept(); err == nil {
+ conn.Close()
+ t.Fatal("retried")
+ }
+}
+
+func TestEncryptedTalkGroup(t *testing.T) {
+ v := startV3(t, "")
+ v.throughBundle(t)
+ v.tls(t, v.pki.serverConfig())
+ v.hmac(t)
+ v.serverAndKey(t, []string{"OPUS"})
+ first := append([]byte{}, v.reg...)
+ v.clock.BlockUntilWaiting(t)
+ v.clock.Advance(10 * time.Second)
+ typ, _ := v.refl.readFrame(t)
+ if typ != 1 {
+ t.Fatalf("type %d, want a TCP heartbeat", typ)
+ }
+ gen := v.clock.Generation()
+ v.refl.writeFrame(t, 1, nil)
+ waitClock(t, v.clock, gen)
+ v.clock.Advance(5 * time.Second)
+ again := v.refl.readRawUDP(t)
+ if !bytes.Equal(first, again) {
+ t.Fatal("registration datagram changed while the heartbeat was missing")
+ }
+ v.beat(t)
+
+ v.st.HoldButton()
+ v.st.WriteMicrophone(make([]int16, opus.FrameSamples))
+ counter, plain := openFromStation(t, v.key, v.iv, v.id, v.refl.readRawUDP(t))
+ if counter != 1 || binary.BigEndian.Uint16(plain[:2]) != 101 {
+ t.Fatalf("audio counter %d type %d", counter, binary.BigEndian.Uint16(plain[:2]))
+ }
+ packet, _ := takeBytes(t, plain[2:])
+ dec, err := opus.NewDecoder()
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer dec.Close()
+ if pcm, err := dec.Decode(packet); err != nil || len(pcm) != opus.FrameSamples {
+ t.Fatalf("decode %v len %d", err, len(pcm))
+ }
+ v.st.ReleaseButton()
+ counter, plain = openFromStation(t, v.key, v.iv, v.id, v.refl.readRawUDP(t))
+ if binary.BigEndian.Uint16(plain[:2]) != 102 {
+ t.Fatalf("flush type %d counter %d", binary.BigEndian.Uint16(plain[:2]), counter)
+ }
+
+ 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)
+ }
+ good := sealToStation(t, v.key, v.iv, 5, append(appendU16(nil, 101), appendBytes(nil, pkt)...))
+ bad := append([]byte{}, good...)
+ bad[len(bad)-1] ^= 0xff
+ v.refl.writeRawUDP(t, bad)
+ v.refl.writeRawUDP(t, sealToStation(t, v.key, v.iv, 0, append(appendU16(nil, 101), appendBytes(nil, pkt)...)))
+ v.refl.writeRawUDP(t, good)
+ if _, err := v.st.WaitPlayback(context.Background(), opus.FrameSamples); err != nil {
+ t.Fatal(err)
+ }
+ if extra := v.st.TakePlayback(); len(extra) != 0 {
+ t.Fatalf("played %d extra samples", len(extra))
+ }
+}
+
+func mustKey(t *testing.T) *rsa.PrivateKey {
+ t.Helper()
+ key, err := rsa.GenerateKey(rand.Reader, 2048)
+ if err != nil {
+ t.Fatal(err)
+ }
+ return key
+}
+
+func hmacSum(key, msg []byte) []byte {
+ mac := hmac.New(sha1.New, key)
+ mac.Write(msg)
+ return mac.Sum(nil)
+}
diff --git a/station/protocol_select_test.go b/station/protocol_select_test.go
new file mode 100644
index 0000000..2efe1ea
--- /dev/null
+++ b/station/protocol_select_test.go
@@ -0,0 +1,140 @@
+package station_test
+
+import (
+ "context"
+ "net"
+ "os"
+ "testing"
+ "time"
+
+ "gitea.neitzel.de/konrad/funknetz/station"
+)
+
+func TestVersionErrorFallsBackWhenTheMaximumIs2(t *testing.T) {
+ for _, msg := range []string{
+ "Unsupported protocol version 3.0. May be at most 2.0.",
+ "Unsupported protocol version 3.0. May be at most 2.5.",
+ } {
+ 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() { _ = st.Run(ctx) }()
+ refl.accept(t)
+ typ, body := refl.readFrame(t)
+ if typ != 5 || !versionIs(body, 3, 0) {
+ t.Fatalf("type %d body %v", typ, body)
+ }
+ refl.writeFrame(t, 13, appendStr(nil, msg))
+ refl.accept(t)
+ typ, body = refl.readFrame(t)
+ if typ != 5 || !versionIs(body, 2, 0) {
+ t.Fatalf("fallback type %d body %v", typ, body)
+ }
+ })
+ }
+}
+
+func TestVersionErrorBelow2Stops(t *testing.T) {
+ for _, msg := range []string{
+ "Unsupported protocol version 3.0. May be at most 1.9.",
+ "Unsupported protocol version 3.0. May be at most 0.6.",
+ } {
+ 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() { _ = 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" || got.Protocol != "" {
+ t.Fatalf("State = %+v", got)
+ }
+ 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")
+ }
+ })
+ }
+}
+
+func TestDowngradeOtherThan2Stops(t *testing.T) {
+ for _, tc := range []struct {
+ major, minor uint16
+ want string
+ }{
+ {1, 0, "Unsupported protocol version 1.0"},
+ {4, 2, "Unsupported protocol version 4.2"},
+ } {
+ t.Run(tc.want, 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() { _ = st.Run(ctx) }()
+ refl.accept(t)
+ refl.readFrame(t)
+ refl.writeFrame(t, 6, versionWire(tc.major, tc.minor))
+ got, err := st.Wait(ctx, func(s station.State) bool { return s.Detail == tc.want })
+ if err != nil {
+ t.Fatal(err)
+ }
+ if got.Protocol != "" || got.Connection != "Disconnected" {
+ t.Fatalf("State = %+v", got)
+ }
+ 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")
+ }
+ })
+ }
+}
+
+func TestDowngradeDoesNotCreateTheCertificateDirectory(t *testing.T) {
+ refl := newReflector(t)
+ dir := t.TempDir()
+ path := dir + "/funknetz.json"
+ body := configWithDir(t, "127.0.0.1", refl.port, "DL1ABC", "key", 20, "./funknetz", "")
+ if err := os.WriteFile(path, []byte(body), 0o600); err != nil {
+ t.Fatal(err)
+ }
+ 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)
+ declineTo2(t, refl)
+ if _, err := os.Stat(dir + "/funknetz"); err == nil {
+ t.Fatal("created the certificate directory on a 2.0 fallback")
+ }
+}
diff --git a/station/reflector_test.go b/station/reflector_test.go
index 286b83f..0750bb0 100644
--- a/station/reflector_test.go
+++ b/station/reflector_test.go
@@ -1,6 +1,7 @@
package station_test
import (
+ "crypto/tls"
"encoding/binary"
"encoding/json"
"io"
@@ -133,10 +134,18 @@ type reflector struct {
ln net.Listener
udp *net.UDPConn
tcp net.Conn
+ tls *tls.Conn
peer *net.UDPAddr
port int
}
+func (r *reflector) conn() net.Conn {
+ if r.tls != nil {
+ return r.tls
+ }
+ return r.tcp
+}
+
func newReflector(t *testing.T) *reflector {
t.Helper()
ln, err := net.Listen("tcp", "127.0.0.1:0")
@@ -166,16 +175,18 @@ func (r *reflector) accept(t *testing.T) {
t.Fatal(err)
}
r.tcp = conn
+ r.tls = nil
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 {
+ conn := r.conn()
+ if err := conn.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 {
+ if _, err := io.ReadFull(conn, hdr[:]); err != nil {
t.Fatal(err)
}
n := binary.BigEndian.Uint32(hdr[:])
@@ -183,7 +194,7 @@ func (r *reflector) readFrame(t *testing.T) (uint16, []byte) {
t.Fatalf("frame length %d", n)
}
buf := make([]byte, n)
- if _, err := io.ReadFull(r.tcp, buf); err != nil {
+ if _, err := io.ReadFull(conn, buf); err != nil {
t.Fatal(err)
}
if len(buf) < 2 {
@@ -199,14 +210,29 @@ func (r *reflector) writeFrame(t *testing.T, typ uint16, body []byte) {
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 {
+ conn := r.conn()
+ if err := conn.SetWriteDeadline(time.Now().Add(2 * time.Second)); err != nil {
t.Fatal(err)
}
- if _, err := r.tcp.Write(append(hdr[:], payload...)); err != nil {
+ if _, err := conn.Write(append(hdr[:], payload...)); err != nil {
t.Fatal(err)
}
}
+func (r *reflector) handshake(t *testing.T, cfg *tls.Config) error {
+ t.Helper()
+ r.tls = tls.Server(r.tcp, cfg)
+ if err := r.tls.SetDeadline(time.Now().Add(2 * time.Second)); err != nil {
+ return err
+ }
+ err := r.tls.Handshake()
+ _ = r.tls.SetDeadline(time.Time{})
+ if err != nil {
+ r.tls = nil
+ }
+ return 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 {
@@ -228,6 +254,33 @@ func (r *reflector) readUDP(t *testing.T) (typ, id, seq uint16, body []byte) {
return typ, id, seq, body
}
+func (r *reflector) readRawUDP(t *testing.T) []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
+ return append([]byte(nil), buf[:n]...)
+}
+
+func (r *reflector) writeRawUDP(t *testing.T, pkt []byte) {
+ t.Helper()
+ if r.peer == nil {
+ t.Fatal("no udp peer")
+ }
+ if err := r.udp.SetWriteDeadline(time.Now().Add(2 * time.Second)); err != nil {
+ t.Fatal(err)
+ }
+ if _, err := r.udp.WriteToUDP(pkt, r.peer); err != nil {
+ t.Fatal(err)
+ }
+}
+
func (r *reflector) writeUDP(t *testing.T, typ, id, seq uint16, body []byte) {
t.Helper()
if r.peer == nil {
@@ -294,8 +347,11 @@ func requireSoftwareIdentity(t *testing.T, body []byte) {
if err := json.Unmarshal([]byte(raw), &got); err != nil {
t.Fatalf("software identity %q: %v", raw, err)
}
- if len(got) != 2 || got["sw"] != station.SoftwareName || got["swVer"] != station.SoftwareVersion {
- t.Fatalf("software identity = %#v, want sw %q and swVer %q only", got, station.SoftwareName, station.SoftwareVersion)
+ if len(got) != 3 || got["sw"] != station.SoftwareName || got["swVer"] != station.SoftwareVersion || got["projVer"] != station.SoftwareVersion {
+ t.Fatalf("software identity = %#v, want sw, swVer, and projVer %q", got, station.SoftwareName)
+ }
+ if _, ok := got["machineArch"]; ok {
+ t.Fatal("protocol 2.0 node information includes machineArch")
}
}
diff --git a/station/session.go b/station/session.go
index 8e02e95..e49f41d 100644
--- a/station/session.go
+++ b/station/session.go
@@ -1,14 +1,21 @@
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"
@@ -16,11 +23,16 @@ import (
const noOpus = "The reflector did not offer OPUS."
-var errStopped = errors.New("session stopped")
+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
@@ -41,6 +53,7 @@ func (s *Station) Run(ctx context.Context) error {
if ctx.Err() != nil {
return nil
}
+ major = 3
s.mu.Lock()
s.noteDownLocked(s.clock.Now(), true)
s.mu.Unlock()
@@ -50,12 +63,30 @@ func (s *Station) Run(ctx context.Context) error {
continue
}
- err = s.session(ctx, conn)
+ 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 {
@@ -63,14 +94,11 @@ func (s *Station) Run(ctx context.Context) error {
}
fatal := s.fatal != ""
s.mu.Unlock()
+ major = 3
if ctx.Err() != nil {
return nil
}
- if fatal {
- <-ctx.Done()
- return nil
- }
- if err != nil && errors.Is(err, errStopped) {
+ if fatal || errors.Is(err, errStopped) {
<-ctx.Done()
return nil
}
@@ -97,7 +125,110 @@ func (s *Station) wait(ctx context.Context, d time.Duration) bool {
}
}
-func (s *Station) session(ctx context.Context, conn net.Conn) error {
+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) }
+
+func (f *framed) startTLS(ctx context.Context, cfg *tls.Config) error {
+ select {
+ case f.upgrade <- cfg:
+ case <-ctx.Done():
+ return ctx.Err()
+ }
+ _ = f.conn.SetReadDeadline(time.Now())
+ select {
+ case err := <-f.up:
+ return err
+ case <-ctx.Done():
+ return ctx.Err()
+ }
+}
+
+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() }) }
@@ -110,38 +241,63 @@ func (s *Station) session(ctx context.Context, conn net.Conn) error {
}
}()
- tcpCh := make(chan frame, 8)
- tcpErr := make(chan error, 1)
- go readTCP(conn, tcpCh, tcpErr, 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(conn, tcpProtoVer, []byte{0, 2, 0, 0}); err != nil {
+ if err := writeFrame(wire, tcpProtoVer, versionWire(major, 0)); err != nil {
return err
}
- lastTCP := s.clock.Now()
- tcpRx := lastTCP.Add(15 * time.Second)
+ 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 udpRx time.Time
+ 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 tcpRx.Before(next) {
- next = tcpRx
+ if t := tcpRx; t.Before(next) {
+ next = t
+ }
+ if t := udpRx; t.Before(next) {
+ next = t
}
if udpReady {
- if u := lastUDP.Add(15 * time.Second); u.Before(next) {
- next = u
- }
- if udpRx.Before(next) {
- next = udpRx
+ 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
@@ -162,13 +318,13 @@ func (s *Station) session(ctx context.Context, conn net.Conn) error {
case <-ctx.Done():
timer.Stop()
return ctx.Err()
- case err := <-tcpErr:
+ case err := <-wire.errs:
timer.Stop()
return err
case err := <-udpFail:
timer.Stop()
return err
- case tcpf = <-tcpCh:
+ case tcpf = <-wire.frames:
timer.Stop()
reason = "tcp"
case udpp = <-udpRecv:
@@ -176,7 +332,7 @@ func (s *Station) session(ctx context.Context, conn net.Conn) error {
reason = "udp"
case <-timer.C():
select {
- case tcpf = <-tcpCh:
+ case tcpf = <-wire.frames:
reason = "tcp"
default:
select {
@@ -188,36 +344,43 @@ func (s *Station) session(ctx context.Context, conn net.Conn) error {
}
}
- now := s.clock.Now()
+ 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 {
+ err := s.onFrame(ctx, wire, major, tcpf, &ph, &authed, &client, &haveCert, done, &udpCh, &udpErr, &udpReady, &lastTCP)
+ if err != nil {
return err
}
case "udp":
- s.mu.Lock()
- if clientIDOf(udpp) == s.clientID {
- udpRx = now.Add(60 * time.Second)
- s.onDatagram(udpp)
+ if err := s.onUDP(wire, udpp, &ph, &udpRx, now, &lastTCP); err != nil {
+ return err
}
- s.mu.Unlock()
default:
+ s.mu.Lock()
+ s.onClock(now)
+ s.mu.Unlock()
if !now.Before(tcpRx) {
- return errors.New("tcp receive timeout")
+ return s.retry("tcp receive timeout")
}
- if udpReady && !now.Before(udpRx) {
- return errors.New("udp receive timeout")
+ if !now.Before(udpRx) {
+ return s.retry("udp receive timeout")
}
if !now.Before(lastTCP.Add(10 * time.Second)) {
- if err := writeFrame(conn, tcpHeartbeat, nil); err != nil {
+ 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)) {
- if err := s.sendUDP(udpHeartbeat, nil); err != nil {
+ var err error
+ if s.udpWait {
+ err = s.sendRegistration()
+ } else {
+ err = s.sendUDP(udpHeartbeat, nil)
+ }
+ if err != nil {
s.mu.Unlock()
return err
}
@@ -227,99 +390,336 @@ func (s *Station) session(ctx context.Context, conn net.Conn) error {
}
}
-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
+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
}
- select {
- case out <- f:
- case <-done:
- return
+ }
+ 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 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 {
+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) 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()
+ case tcpHeartbeat:
return nil
case tcpError:
- msg, _, err := parseString(f.body)
- if err != nil {
- msg = "could not be read"
- }
- s.endFatal(msg, true)
- return errStopped
+ 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)
+}
+
+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:
- return s.onServerInfo(conn, f.body, done, udpCh, udpErr, udpReady, udpRx, lastTCP)
+ 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) 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 {
+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
}
- if !hasOPUS(codecs) {
- s.endFatal(noOpus, true)
+ *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) 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)
+ 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
}
- enc, err := opus.NewEncoder()
+ 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) 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 := wire.startTLS(ctx, cfg); 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
}
- dec, err := opus.NewDecoder()
- if err != nil {
- enc.Close()
- return err
+ if enc == nil {
+ return errStopped
}
info, err := softwareIdentityBody()
if err != nil {
@@ -327,27 +727,20 @@ func (s *Station) onServerInfo(conn net.Conn, body []byte, done <-chan struct{},
dec.Close()
return err
}
- if err := writeFrame(conn, tcpNodeInfo, info); err != nil {
+ 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(conn, tcpSelectTG, tg[:]); err != nil {
+ if err := writeFrame(wire, 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})
+ udpConn, peer, err := dialAudio(wire.conn)
if err != nil {
enc.Close()
dec.Close()
@@ -355,14 +748,16 @@ func (s *Station) onServerInfo(conn net.Conn, body []byte, done <-chan struct{},
}
ch := make(chan []byte, 32)
errc := make(chan error, 1)
- go readUDP(udpConn, ch, errc, done)
-
+ 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()
@@ -378,11 +773,311 @@ func (s *Station) onServerInfo(conn net.Conn, body []byte, done <-chan struct{},
s.writeLog(s.clock.Now(), "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) 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
}
@@ -425,69 +1120,21 @@ func (s *Station) onTalker(body []byte, start bool) {
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 {
- // A lost datagram ends the session when the receive timer expires.
_ = s.sendUDP(udpAudio, appendBlob(nil, pkt))
}
} else {
s.mic = nil
}
if s.udp != nil {
- // A lost datagram ends the session when the receive timer expires.
_ = s.sendUDP(udpFlush, nil)
+ s.flushWaiting = true
+ s.flushAt = s.clock.Now()
}
}
@@ -496,6 +1143,7 @@ func (s *Station) closeUDP() {
s.udp.Close()
s.udp = nil
}
+ s.udpPeer = nil
if s.enc != nil {
s.enc.Close()
s.enc = nil
@@ -522,8 +1170,8 @@ func (s *Station) publish() {
s.state.Detail = s.fatal
s.state.Talker = ""
s.state.Transmitting = false
+ s.state.Protocol = ""
} else {
- s.state.Detail = "Idle"
tx := s.transmitting()
s.state.Transmitting = tx
if tx {
@@ -532,8 +1180,22 @@ func (s *Station) publish() {
} else {
s.state.Talker = s.remoteTalker
}
- if s.state.Talker == "" && s.logNotice != "" {
- s.state.Detail = s.logNotice
+ 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 {
@@ -541,3 +1203,45 @@ func (s *Station) publish() {
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
+}
diff --git a/station/session_test.go b/station/session_test.go
index 1c7d855..8813afb 100644
--- a/station/session_test.go
+++ b/station/session_test.go
@@ -27,14 +27,20 @@ func TestLoginAnnouncesProtocolAndJoinsTheTalkGroup(t *testing.T) {
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)
+ if got.Callsign != "DL1ABC" || got.TalkGroup != 20 || got.Detail != "Idle" || got.Protocol != "" {
+ t.Fatalf("State = %+v, want DL1ABC talk group 20 Idle and a blank protocol", 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)
+ if typ != 5 || !versionIs(body, 3, 0) {
+ t.Fatalf("first frame type %d body %v, want protocol 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 frame type %d body %v, want protocol 2.0", typ, body)
}
challenge := make([]byte, 20)
@@ -91,8 +97,8 @@ func TestLoginAnnouncesProtocolAndJoinsTheTalkGroup(t *testing.T) {
if err != nil {
t.Fatal(err)
}
- if got.Detail != "Idle" || got.Talker != "" {
- t.Fatalf("State = %+v, want Connected Idle", got)
+ if got.Detail != "Idle" || got.Talker != "" || got.Protocol != "2.0" {
+ t.Fatalf("State = %+v, want Connected Idle protocol 2.0", got)
}
// No other talk group: the next TCP message is a heartbeat, not a monitor set.
diff --git a/station/software.go b/station/software.go
index d772f87..ee70483 100644
--- a/station/software.go
+++ b/station/software.go
@@ -1,6 +1,11 @@
package station
-import jsonv2 "encoding/json/v2"
+import (
+ "bytes"
+
+ jsonv2 "encoding/json/v2"
+ "golang.org/x/sys/unix"
+)
// SoftwareName and SoftwareVersion are the identity sent to the reflector
// after login. Edit them here.
@@ -9,16 +14,56 @@ const (
SoftwareVersion = "1.0.0"
)
+type nodeSoftware struct {
+ SW string `json:"sw"`
+ SWVer string `json:"swVer"`
+ ProjVer string `json:"projVer"`
+ MachineArch string `json:"machineArch,omitempty"`
+}
+
func softwareIdentityBody() ([]byte, error) {
- raw, err := jsonv2.Marshal(struct {
- SW string `json:"sw"`
- SWVer string `json:"swVer"`
- }{
- SW: SoftwareName,
- SWVer: SoftwareVersion,
- })
+ raw, err := marshalNode(false)
if err != nil {
return nil, err
}
return appendString(nil, string(raw)), nil
}
+
+func nodeInfoV3(iv, key []byte) ([]byte, error) {
+ raw, err := marshalNode(true)
+ if err != nil {
+ return nil, err
+ }
+ body := appendBlob(nil, iv)
+ body = appendBlob(body, key)
+ return appendString(body, string(raw)), nil
+}
+
+func marshalNode(arch bool) ([]byte, error) {
+ info := nodeSoftware{
+ SW: SoftwareName,
+ SWVer: SoftwareVersion,
+ ProjVer: SoftwareVersion,
+ }
+ if arch {
+ if name, ok := machineArch(); ok {
+ info.MachineArch = name
+ }
+ }
+ return jsonv2.Marshal(info)
+}
+
+func machineArch() (string, bool) {
+ var u unix.Utsname
+ if err := unix.Uname(&u); err != nil {
+ return "", false
+ }
+ b := u.Machine[:]
+ if i := bytes.IndexByte(b, 0); i >= 0 {
+ b = b[:i]
+ }
+ if len(b) == 0 {
+ return "", false
+ }
+ return string(b), true
+}
diff --git a/station/station.go b/station/station.go
index 6cac3a4..920ea84 100644
--- a/station/station.go
+++ b/station/station.go
@@ -24,6 +24,7 @@ type State struct {
Detail string `json:"detail"`
Callsign string `json:"callsign"`
TalkGroup int `json:"talkGroup"`
+ Protocol string `json:"protocol"`
Talker string `json:"talker"`
Transmitting bool `json:"transmitting"`
}
@@ -39,13 +40,27 @@ type Station struct {
buttonHeld bool
spaceHeld bool
fatal string
+ retryReason string
+ protocol string
remoteTalker string
roomBusy bool
+ audioHeard bool
+ lastAudio time.Time
+ flushWaiting bool
+ flushAt time.Time
+ rxSeq uint16
+ secure bool
+ udpWait bool
+ ivPrefix []byte
+ udpKey []byte
+ rxCounter uint32
+ txCounter uint32
mic []int16
play []int16
clientID uint16
seq uint16
udp *net.UDPConn
+ udpPeer *net.UDPAddr
lastUDPSend time.Time
enc *opus.Encoder
dec *opus.Decoder
@@ -73,7 +88,9 @@ func New(opt Options) (*Station, error) {
clock = wallClock{}
}
detail := "Idle"
- if cfg.logNotice != "" {
+ if cfg.certFatal != "" {
+ detail = cfg.certFatal
+ } else if cfg.logNotice != "" {
detail = cfg.logNotice
}
s := &Station{
@@ -82,6 +99,7 @@ func New(opt Options) (*Station, error) {
wake: make(chan struct{}, 1),
logPath: cfg.logPath,
logNotice: cfg.logNotice,
+ fatal: cfg.certFatal,
state: State{
Connection: "Disconnected",
Detail: detail,
diff --git a/station/tls.go b/station/tls.go
new file mode 100644
index 0000000..bf4bfe4
--- /dev/null
+++ b/station/tls.go
@@ -0,0 +1,60 @@
+package station
+
+import (
+ "crypto/tls"
+ "crypto/x509"
+ "errors"
+ "net"
+ "strings"
+)
+
+var errHostMismatch = errors.New("host mismatch")
+
+func tlsConfig(host string, caPEM []byte, client tls.Certificate, present bool) (*tls.Config, error) {
+ pool := x509.NewCertPool()
+ if !pool.AppendCertsFromPEM(caPEM) {
+ return nil, errors.New("ca bundle")
+ }
+ cfg := &tls.Config{
+ MinVersion: tls.VersionTLS12,
+ RootCAs: pool,
+ InsecureSkipVerify: true,
+ VerifyConnection: func(cs tls.ConnectionState) error {
+ return verifyPeer(host, pool, cs)
+ },
+ }
+ if present {
+ cfg.Certificates = []tls.Certificate{client}
+ }
+ return cfg, nil
+}
+
+func verifyPeer(host string, pool *x509.CertPool, cs tls.ConnectionState) error {
+ if len(cs.PeerCertificates) == 0 {
+ return errors.New("no peer certificate")
+ }
+ leaf := cs.PeerCertificates[0]
+ opts := x509.VerifyOptions{Roots: pool, KeyUsages: []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth}}
+ if len(cs.PeerCertificates) > 1 {
+ opts.Intermediates = x509.NewCertPool()
+ for _, cert := range cs.PeerCertificates[1:] {
+ opts.Intermediates.AddCert(cert)
+ }
+ }
+ if _, err := leaf.Verify(opts); err != nil {
+ return err
+ }
+ host = strings.TrimSuffix(host, ".")
+ if ip := net.ParseIP(host); ip != nil {
+ for _, cand := range leaf.IPAddresses {
+ if cand.Equal(ip) {
+ return nil
+ }
+ }
+ return errHostMismatch
+ }
+ if err := leaf.VerifyHostname(host); err != nil {
+ return errHostMismatch
+ }
+ return nil
+}