From e642b24be00e7245514e9e07023cee4f9ace5fcd Mon Sep 17 00:00:00 2001 From: Konrad Neitzel Date: Wed, 7 Oct 2026 15:01:38 +0200 Subject: [PATCH] Separate the window sentence from the wrapped error so a bad certificate file is not reported as a key mismatch. Co-authored-by: Cursor --- internal/gcm64/gcm.go | 63 +++ station/behavior_test.go | 20 +- station/cert.go | 87 +++-- station/cipher.go | 60 +-- station/config_test.go | 5 +- station/lab_reflector_test.go | 5 +- station/log.go | 10 +- station/log_test.go | 10 +- station/protocol2_test.go | 15 +- station/protocol3_test.go | 93 ++--- station/protocol_select_test.go | 20 +- station/reflector_test.go | 1 + station/session.go | 664 +++++--------------------------- station/session_test.go | 5 +- station/session_udp.go | 246 ++++++++++++ station/session_v3.go | 259 +++++++++++++ station/software.go | 21 +- 17 files changed, 842 insertions(+), 742 deletions(-) create mode 100644 internal/gcm64/gcm.go create mode 100644 station/session_udp.go create mode 100644 station/session_v3.go diff --git a/internal/gcm64/gcm.go b/internal/gcm64/gcm.go new file mode 100644 index 0000000..f146a9c --- /dev/null +++ b/internal/gcm64/gcm.go @@ -0,0 +1,63 @@ +// Package gcm64 seals and opens AES-GCM with the 8-byte tag this protocol uses. +package gcm64 + +import ( + "crypto/aes" + "crypto/cipher" + "crypto/subtle" + "encoding/binary" + "errors" +) + +// IV is the 12-byte nonce: a 6-byte prefix, the client id, and the counter. +func IV(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 +} + +// Seal returns an 8-byte tag and the ciphertext. The tag is the leading 8 +// bytes of the ordinary 16-byte GCM tag, because crypto/cipher only builds +// tags of 12 to 16 bytes. +func Seal(key, nonce, plain, aad []byte) (tag, ct []byte, err error) { + g, err := full(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 +} + +// Open checks the 8-byte tag and returns the plaintext. +func Open(key, nonce, tag, ct, aad []byte) ([]byte, error) { + if len(tag) != 8 { + return nil, errors.New("tag") + } + g, err := full(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 full(key []byte) (cipher.AEAD, error) { + block, err := aes.NewCipher(key) + if err != nil { + return nil, err + } + return cipher.NewGCM(block) +} diff --git a/station/behavior_test.go b/station/behavior_test.go index 5d99692..eb9663e 100644 --- a/station/behavior_test.go +++ b/station/behavior_test.go @@ -43,7 +43,10 @@ func loginLog(t *testing.T, callsign, key string, tg int, id uint16, codecs []st } ctx, cancel := context.WithCancel(t.Context()) t.Cleanup(cancel) - go func() { _ = st.Run(ctx) }() + go func() { + // Run's error is observed through State, which is the station seam. + _ = st.Run(ctx) + }() refl.accept(t) declineTo2(t, refl) challenge := make([]byte, 20) @@ -187,7 +190,10 @@ func TestRefusedConnectionRetriesUntilQuit(t *testing.T) { } ctx, cancel := context.WithCancel(t.Context()) defer cancel() - go func() { _ = st.Run(ctx) }() + go func() { + // Run's error is observed through State, which is the station seam. + _ = st.Run(ctx) + }() if _, err := st.Wait(ctx, func(s station.State) bool { return s.Connection == "Disconnected" }); err != nil { t.Fatal(err) @@ -232,7 +238,10 @@ func TestReflectorErrorStopsRetries(t *testing.T) { } ctx, cancel := context.WithCancel(t.Context()) defer cancel() - go func() { _ = st.Run(ctx) }() + go func() { + // Run's error is observed through State, which is the station seam. + _ = st.Run(ctx) + }() refl.accept(t) refl.readFrame(t) refl.writeFrame(t, 13, appendStr(nil, msg)) @@ -266,7 +275,10 @@ func TestMissingOpusStopsRetries(t *testing.T) { } ctx, cancel := context.WithCancel(t.Context()) defer cancel() - go func() { _ = st.Run(ctx) }() + go func() { + // Run's error is observed through State, which is the station seam. + _ = st.Run(ctx) + }() refl.accept(t) declineTo2(t, refl) challenge := make([]byte, 20) diff --git a/station/cert.go b/station/cert.go index 7017f4f..f173bd2 100644 --- a/station/cert.go +++ b/station/cert.go @@ -11,6 +11,7 @@ import ( "encoding/asn1" "encoding/pem" "errors" + "fmt" "io/fs" "os" "path/filepath" @@ -84,13 +85,24 @@ 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 +func csrNewerThanCert(dir string) (bool, error) { + csrPath := filepath.Join(dir, csrFileName) + certPath := filepath.Join(dir, certFileName) + csr, err := os.Stat(csrPath) + if err != nil && !errors.Is(err, fs.ErrNotExist) { + return false, fmt.Errorf("stat %s: %w", csrPath, err) } - return csr.ModTime().After(cert.ModTime()) + if err != nil { + return false, nil + } + cert, err := os.Stat(certPath) + if err != nil && !errors.Is(err, fs.ErrNotExist) { + return false, fmt.Errorf("stat %s: %w", certPath, err) + } + if err != nil { + return false, nil + } + return csr.ModTime().After(cert.ModTime()), nil } func readCSR(dir string) (string, error) { @@ -103,7 +115,7 @@ func readCSR(dir string) (string, error) { // 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) { +func prepareIdentity(dir, callsign string) (tls.Certificate, bool, string, error) { certPath := filepath.Join(dir, certFileName) keyPath := filepath.Join(dir, keyFileName) _, certErr := os.Stat(certPath) @@ -111,100 +123,103 @@ func prepareIdentity(dir, callsign string) (tls.Certificate, bool, string) { certExists := certErr == nil keyExists := keyErr == nil if certErr != nil && !errors.Is(certErr, fs.ErrNotExist) { - return tls.Certificate{}, false, "Cannot create " + certPath + return tls.Certificate{}, false, "Cannot create " + certPath, fmt.Errorf("stat certificate: %w", certErr) } if keyErr != nil && !errors.Is(keyErr, fs.ErrNotExist) { - return tls.Certificate{}, false, "Cannot create " + keyPath + return tls.Certificate{}, false, "Cannot create " + keyPath, fmt.Errorf("stat key: %w", keyErr) } if certExists && !keyExists { - return tls.Certificate{}, false, "Certificate key is missing" + return tls.Certificate{}, false, "Certificate key is missing", nil } if certExists && keyExists { - pair, msg := loadPair(certPath, keyPath, callsign) + pair, msg, err := loadPair(certPath, keyPath, callsign) if msg != "" { - return tls.Certificate{}, false, msg + return tls.Certificate{}, false, msg, err } - if msg := ensureCSR(dir, callsign); msg != "" { - return tls.Certificate{}, false, msg + if msg, err := ensureCSR(dir, callsign); msg != "" { + return tls.Certificate{}, false, msg, err } - return pair, true, "" + return pair, true, "", nil } - if msg := ensureKeyAndCSR(dir, callsign, !keyExists); msg != "" { - return tls.Certificate{}, false, msg + if msg, err := ensureKeyAndCSR(dir, callsign, !keyExists); msg != "" { + return tls.Certificate{}, false, msg, err } - return tls.Certificate{}, false, "" + return tls.Certificate{}, false, "", nil } -func loadPair(certPath, keyPath, callsign string) (tls.Certificate, string) { +func loadPair(certPath, keyPath, callsign string) (tls.Certificate, string, error) { cert, err := loadCert(certPath) if err != nil { - return tls.Certificate{}, "Certificate does not match the key" + return tls.Certificate{}, "Cannot create " + certPath, fmt.Errorf("read certificate: %w", err) } key, err := loadKey(keyPath) if err != nil { - return tls.Certificate{}, "Certificate does not match the key" + return tls.Certificate{}, "Cannot create " + keyPath, fmt.Errorf("read key: %w", err) } pub, ok := cert.PublicKey.(*rsa.PublicKey) if !ok || !pub.Equal(&key.PublicKey) { - return tls.Certificate{}, "Certificate does not match the key" + return tls.Certificate{}, "Certificate does not match the key", nil } if cert.Subject.CommonName != callsign { - return tls.Certificate{}, "Certificate callsign does not match" + return tls.Certificate{}, "Certificate callsign does not match", nil } return tls.Certificate{ Certificate: [][]byte{cert.Raw}, PrivateKey: key, Leaf: cert, - }, "" + }, "", nil } -func ensureCSR(dir, callsign string) string { +func ensureCSR(dir, callsign string) (string, error) { csrPath := filepath.Join(dir, csrFileName) _, err := os.Stat(csrPath) if err == nil { - return "" + return "", nil } if !errors.Is(err, fs.ErrNotExist) { - return "Cannot create " + csrPath + return "Cannot create " + csrPath, fmt.Errorf("stat signing request: %w", err) } key, err := loadKey(filepath.Join(dir, keyFileName)) if err != nil { - return "Cannot create " + csrPath + return "Cannot create " + csrPath, fmt.Errorf("read key: %w", err) } if err := writeCSR(csrPath, key, callsign); err != nil { - return "Cannot create " + csrPath + return "Cannot create " + csrPath, err } - return "" + return "", nil } -func ensureKeyAndCSR(dir, callsign string, created bool) string { +func ensureKeyAndCSR(dir, callsign string, created bool) (string, error) { 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 + return "Cannot create " + keyPath, fmt.Errorf("generate key: %w", err) } 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 + return "Cannot create " + keyPath, err } } else { key, err = loadKey(keyPath) if err != nil { - return "Cannot create " + keyPath + return "Cannot create " + keyPath, fmt.Errorf("read key: %w", err) } } csrPath := filepath.Join(dir, csrFileName) _, csrErr := os.Stat(csrPath) + if csrErr != nil && !errors.Is(csrErr, fs.ErrNotExist) { + return "Cannot create " + csrPath, fmt.Errorf("stat signing request: %w", csrErr) + } if created || errors.Is(csrErr, fs.ErrNotExist) { if err := writeCSR(csrPath, key, callsign); err != nil { - return "Cannot create " + csrPath + return "Cannot create " + csrPath, err } } - return "" + return "", nil } func writeCSR(path string, key *rsa.PrivateKey, callsign string) error { diff --git a/station/cipher.go b/station/cipher.go index 8c0eab4..a9a76eb 100644 --- a/station/cipher.go +++ b/station/cipher.go @@ -1,11 +1,9 @@ package station import ( - "crypto/aes" - "crypto/cipher" - "crypto/subtle" "encoding/binary" - "errors" + + "gitea.neitzel.de/konrad/funknetz/internal/gcm64" ) func counterAAD(counter uint32) []byte { @@ -19,16 +17,8 @@ func registrationAAD(clientID uint16) []byte { 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) + tag, ct, err := gcm64.Seal(key, gcm64.IV(prefix, clientID, counter), plain, aad) if err != nil { return nil, err } @@ -39,47 +29,5 @@ func sealDatagram(key, prefix []byte, clientID uint16, counter uint32, aad, plai } 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) + return gcm64.Open(key, gcm64.IV(prefix, clientID, counter), tag, ct, aad) } diff --git a/station/config_test.go b/station/config_test.go index 0bfef63..e33614b 100644 --- a/station/config_test.go +++ b/station/config_test.go @@ -467,7 +467,10 @@ func TestCertificateDirStopsBeforeASocket(t *testing.T) { } ctx, cancel := context.WithCancel(t.Context()) defer cancel() - go func() { _ = st.Run(ctx) }() + go func() { + // Run's error is observed through State, which is the station seam. + _ = st.Run(ctx) + }() ln.(*net.TCPListener).SetDeadline(time.Now().Add(200 * time.Millisecond)) if conn, err := ln.Accept(); err == nil { conn.Close() diff --git a/station/lab_reflector_test.go b/station/lab_reflector_test.go index 888b638..00c9a81 100644 --- a/station/lab_reflector_test.go +++ b/station/lab_reflector_test.go @@ -44,7 +44,10 @@ func TestConnectsToLabReflector(t *testing.T) { ctx, cancel := context.WithTimeout(t.Context(), 15*time.Second) defer cancel() - go func() { _ = st.Run(ctx) }() + go func() { + // Run's error is observed through State, which is the station seam. + _ = st.Run(ctx) + }() state, err := st.Wait(ctx, func(s station.State) bool { return s.Connection == "Connected" diff --git a/station/log.go b/station/log.go index 1a487a6..1f592b0 100644 --- a/station/log.go +++ b/station/log.go @@ -93,6 +93,10 @@ func (s *Station) noteDownLocked(now time.Time, dial bool) { } func (s *Station) endFatal(msg string, log bool) { + s.endFatalCause(msg, nil, log) +} + +func (s *Station) endFatalCause(msg string, cause error, log bool) { s.mu.Lock() defer s.mu.Unlock() if s.fatal != "" { @@ -107,7 +111,11 @@ func (s *Station) endFatal(msg string, log bool) { if log { s.up = false s.outage = true - s.writeLog(now, oneLine(msg)) + line := oneLine(msg) + if cause != nil { + line += ": " + oneLine(cause.Error()) + } + s.writeLog(now, line) } s.publish() } diff --git a/station/log_test.go b/station/log_test.go index a34f670..ed72deb 100644 --- a/station/log_test.go +++ b/station/log_test.go @@ -131,7 +131,10 @@ func TestFirstDialFailureIsLoggedOnce(t *testing.T) { } ctx, cancel := context.WithCancel(t.Context()) defer cancel() - go func() { _ = st.Run(ctx) }() + go func() { + // Run's error is observed through State, which is the station seam. + _ = st.Run(ctx) + }() want := "2026-10-03T12:00:00Z disconnected\n" deadline := time.Now().Add(2 * time.Second) @@ -160,7 +163,10 @@ func TestReflectorErrorsAreOneLogLine(t *testing.T) { } ctx, cancel := context.WithCancel(t.Context()) defer cancel() - go func() { _ = st.Run(ctx) }() + go func() { + // Run's error is observed through State, which is the station seam. + _ = st.Run(ctx) + }() refl.accept(t) refl.readFrame(t) refl.writeFrame(t, 13, appendStr(nil, "bad\nline")) diff --git a/station/protocol2_test.go b/station/protocol2_test.go index b62d333..75e70ca 100644 --- a/station/protocol2_test.go +++ b/station/protocol2_test.go @@ -120,7 +120,10 @@ func TestUDPLimitRunsBeforeAnyDatagram(t *testing.T) { } ctx, cancel := context.WithCancel(t.Context()) defer cancel() - go func() { _ = st.Run(ctx) }() + go func() { + // Run's error is observed through State, which is the station seam. + _ = st.Run(ctx) + }() refl.accept(t) declineTo2(t, refl) for step := 0; step < 6; step++ { @@ -156,7 +159,10 @@ func TestOversizedFrameBeforeAuthRetries(t *testing.T) { } ctx, cancel := context.WithCancel(t.Context()) defer cancel() - go func() { _ = st.Run(ctx) }() + go func() { + // Run's error is observed through State, which is the station seam. + _ = st.Run(ctx) + }() refl.accept(t) declineTo2(t, refl) payload := make([]byte, 65) @@ -187,7 +193,10 @@ func TestApplicationMessageBeforeLoginRetries(t *testing.T) { } ctx, cancel := context.WithCancel(t.Context()) defer cancel() - go func() { _ = st.Run(ctx) }() + go func() { + // Run's error is observed through State, which is the station seam. + _ = st.Run(ctx) + }() refl.accept(t) declineTo2(t, refl) refl.writeFrame(t, 100, serverInfo(1, nil, []string{"OPUS"})) diff --git a/station/protocol3_test.go b/station/protocol3_test.go index 3c45047..154e463 100644 --- a/station/protocol3_test.go +++ b/station/protocol3_test.go @@ -4,8 +4,6 @@ import ( "bytes" "context" "crypto" - "crypto/aes" - "crypto/cipher" "crypto/hmac" "crypto/rand" "crypto/rsa" @@ -16,9 +14,10 @@ import ( "crypto/x509/pkix" "encoding/asn1" "encoding/binary" - "encoding/json" + jsonv2 "encoding/json/v2" "encoding/pem" "errors" + "io/fs" "math/big" "net" "os" @@ -26,9 +25,7 @@ import ( "testing" "time" - jsonv2 "encoding/json/v2" - "golang.org/x/sys/unix" - + "gitea.neitzel.de/konrad/funknetz/internal/gcm64" "gitea.neitzel.de/konrad/funknetz/internal/opus" "gitea.neitzel.de/konrad/funknetz/station" ) @@ -194,7 +191,10 @@ func startV3(t *testing.T, certDir string) *v3 { } ctx, cancel := context.WithCancel(t.Context()) t.Cleanup(cancel) - go func() { _ = st.Run(ctx) }() + go func() { + // Run's error is observed through State, which is the station seam. + _ = 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} } @@ -276,21 +276,17 @@ func (v *v3) serverAndKey(t *testing.T, codecs []string) { 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 { + if err := jsonv2.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, ok := station.MachineArch() + if !ok { + t.Fatal("machine arch") } - name := u.Machine[:] - if i := bytes.IndexByte(name, 0); i >= 0 { - name = name[:i] - } - if got["machineArch"] != string(name) { + if got["machineArch"] != name { t.Fatalf("machineArch %q, want %q", got["machineArch"], name) } v.iv = iv @@ -403,7 +399,7 @@ func openReg(t *testing.T, key, iv []byte, id uint16, pkt []byte) []byte { 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) + plain, err := gcm64.Open(key, gcm64.IV(iv, id, counter), tag, ct, aad) if err != nil { t.Fatal(err) } @@ -414,7 +410,7 @@ func sealToStation(t *testing.T, key, iv []byte, counter uint32, plain []byte) [ t.Helper() var aad [4]byte binary.BigEndian.PutUint32(aad[:], counter) - tag, ct, err := sealTruncTest(key, nonceFor(iv, 0, counter), plain, aad[:]) + tag, ct, err := gcm64.Seal(key, gcm64.IV(iv, 0, counter), plain, aad[:]) if err != nil { t.Fatal(err) } @@ -423,48 +419,6 @@ func sealToStation(t *testing.T, key, iv []byte, counter uint32, plain []byte) [ 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 { @@ -505,7 +459,7 @@ func TestBadCASignatureRetriesOn3(t *testing.T) { if got.Connection != "Disconnected" { t.Fatalf("Connection %s", got.Connection) } - if _, err := os.Stat(filepath.Join(v.certDir, "ca.pem")); !os.IsNotExist(err) { + if _, err := os.Stat(filepath.Join(v.certDir, "ca.pem")); !errors.Is(err, fs.ErrNotExist) { t.Fatalf("ca.pem err = %v, want absent", err) } v.clock.BlockUntilWaiting(t) @@ -568,7 +522,10 @@ func TestCADigestMismatchIsLoggedAndKept(t *testing.T) { } ctx, cancel := context.WithCancel(t.Context()) defer cancel() - go func() { _ = st.Run(ctx) }() + go func() { + // Run's error is observed through State, which is the station seam. + _ = st.Run(ctx) + }() pki := newPKI(t, "", net.ParseIP("127.0.0.1")) refl.accept(t) refl.readFrame(t) @@ -607,7 +564,10 @@ func TestCannotCreateCertificateDirStops(t *testing.T) { } ctx, cancel := context.WithCancel(t.Context()) defer cancel() - go func() { _ = st.Run(ctx) }() + go func() { + // Run's error is observed through State, which is the station seam. + _ = st.Run(ctx) + }() pki := newPKI(t, "", net.ParseIP("127.0.0.1")) refl.accept(t) refl.readFrame(t) @@ -645,7 +605,10 @@ func TestRelativeCertificateDirStaysRelative(t *testing.T) { } ctx, cancel := context.WithCancel(t.Context()) defer cancel() - go func() { _ = st.Run(ctx) }() + go func() { + // Run's error is observed through State, which is the station seam. + _ = st.Run(ctx) + }() pki := newPKI(t, "", net.ParseIP("127.0.0.1")) refl.accept(t) refl.readFrame(t) @@ -777,6 +740,7 @@ func TestHostMismatchRetriesOn3(t *testing.T) { v.pki = newPKI(t, "example.com", nil) v.throughBundle(t) v.refl.writeFrame(t, 15, nil) + // The station records this handshake failure; its State is the result. _ = 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" @@ -853,6 +817,7 @@ func TestTLSHandshakeFailureRetriesOn3(t *testing.T) { cfg.MinVersion = tls.VersionTLS10 cfg.MaxVersion = tls.VersionTLS11 v.refl.writeFrame(t, 15, nil) + // The station records this handshake failure; its State is the result. _ = 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) diff --git a/station/protocol_select_test.go b/station/protocol_select_test.go index 2efe1ea..eccefd4 100644 --- a/station/protocol_select_test.go +++ b/station/protocol_select_test.go @@ -25,7 +25,10 @@ func TestVersionErrorFallsBackWhenTheMaximumIs2(t *testing.T) { } ctx, cancel := context.WithCancel(t.Context()) defer cancel() - go func() { _ = st.Run(ctx) }() + go func() { + // Run's error is observed through State, which is the station seam. + _ = st.Run(ctx) + }() refl.accept(t) typ, body := refl.readFrame(t) if typ != 5 || !versionIs(body, 3, 0) { @@ -56,7 +59,10 @@ func TestVersionErrorBelow2Stops(t *testing.T) { } ctx, cancel := context.WithCancel(t.Context()) defer cancel() - go func() { _ = st.Run(ctx) }() + go func() { + // Run's error is observed through State, which is the station seam. + _ = st.Run(ctx) + }() refl.accept(t) refl.readFrame(t) refl.writeFrame(t, 13, appendStr(nil, msg)) @@ -95,7 +101,10 @@ func TestDowngradeOtherThan2Stops(t *testing.T) { } ctx, cancel := context.WithCancel(t.Context()) defer cancel() - go func() { _ = st.Run(ctx) }() + go func() { + // Run's error is observed through State, which is the station seam. + _ = st.Run(ctx) + }() refl.accept(t) refl.readFrame(t) refl.writeFrame(t, 6, versionWire(tc.major, tc.minor)) @@ -131,7 +140,10 @@ func TestDowngradeDoesNotCreateTheCertificateDirectory(t *testing.T) { } ctx, cancel := context.WithCancel(t.Context()) defer cancel() - go func() { _ = st.Run(ctx) }() + go func() { + // Run's error is observed through State, which is the station seam. + _ = st.Run(ctx) + }() refl.accept(t) declineTo2(t, refl) if _, err := os.Stat(dir + "/funknetz"); err == nil { diff --git a/station/reflector_test.go b/station/reflector_test.go index 06a2ead..c243e7d 100644 --- a/station/reflector_test.go +++ b/station/reflector_test.go @@ -236,6 +236,7 @@ func (r *reflector) handshake(t *testing.T, cfg *tls.Config) error { return err } err := r.tls.Handshake() + // Clearing the deadline is best-effort; the handshake error is the result. _ = r.tls.SetDeadline(time.Time{}) if err != nil { r.tls = nil diff --git a/station/session.go b/station/session.go index 176c174..8427511 100644 --- a/station/session.go +++ b/station/session.go @@ -1,19 +1,17 @@ package station import ( - "bytes" "context" "crypto/hmac" - "crypto/rand" "crypto/sha1" "crypto/tls" "encoding/binary" "errors" + "fmt" "io" "net" - "os" - "path/filepath" "strconv" + "strings" "sync" "sync/atomic" "time" @@ -126,27 +124,40 @@ func (s *Station) wait(ctx context.Context, d time.Duration) bool { } 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 + conn net.Conn + mu sync.Mutex + w io.Writer + frames chan frame + errs chan error + upgrade chan *tls.Config + handshake chan error + max atomic.Uint32 } -func newFramed(conn net.Conn, max uint32, done <-chan struct{}) *framed { +// exchange is the state one TCP session's frame handlers share. +type exchange struct { + ph phase + authed bool + client tls.Certificate + haveCert bool + done chan struct{} + udpCh chan []byte + udpErr chan error + udpReady bool + lastTCP time.Time + limit time.Time +} + +func newFramed(conn net.Conn, max uint32) *framed { f := &framed{ - conn: conn, - w: conn, - frames: make(chan frame, 8), - errs: make(chan error, 1), - upgrade: make(chan *tls.Config, 1), - up: make(chan error, 1), + conn: conn, + w: conn, + frames: make(chan frame, 8), + errs: make(chan error, 1), + upgrade: make(chan *tls.Config, 1), + handshake: make(chan error, 1), } f.max.Store(max) - go f.read(done) return f } @@ -167,7 +178,9 @@ func (s *Station) startTLS(ctx context.Context, f *framed, cfg *tls.Config, unti case <-ctx.Done(): return ctx.Err() } - _ = f.conn.SetReadDeadline(time.Now()) + if err := f.conn.SetReadDeadline(time.Now()); err != nil { + return err + } wait := until.Sub(s.clock.Now()) if wait < 0 { wait = 0 @@ -175,15 +188,19 @@ func (s *Station) startTLS(ctx context.Context, f *framed, cfg *tls.Config, unti timer := s.clock.NewTimer(wait) defer timer.Stop() select { - case err := <-f.up: + case err := <-f.handshake: return err case <-ctx.Done(): - _ = f.conn.SetDeadline(time.Now()) + if err := f.conn.SetDeadline(time.Now()); err != nil { + return err + } return ctx.Err() case <-timer.C(): - _ = f.conn.SetDeadline(time.Now()) + if err := f.conn.SetDeadline(time.Now()); err != nil { + return err + } select { - case <-f.up: + case <-f.handshake: case <-ctx.Done(): } return errTLSTimeout @@ -195,22 +212,24 @@ func (f *framed) read(done <-chan struct{}) { for { fr, err := readFrameMax(src, f.max.Load()) if err != nil { - var ne net.Error - if errors.As(err, &ne) && ne.Timeout() { + if ne, ok := errors.AsType[net.Error](err); ok && ne.Timeout() { select { case cfg := <-f.upgrade: - _ = f.conn.SetReadDeadline(time.Time{}) + if derr := f.conn.SetReadDeadline(time.Time{}); derr != nil { + f.handshake <- derr + return + } tc := tls.Client(f.conn, cfg) herr := tc.Handshake() if herr != nil { - f.up <- herr + f.handshake <- herr return } f.mu.Lock() f.w = tc f.mu.Unlock() src = tc - f.up <- nil + f.handshake <- nil continue case <-done: return @@ -257,15 +276,16 @@ func (s *Station) session(ctx context.Context, conn net.Conn, major uint16) erro } }() - limit := uint32(maxFrame) - ph := phCA + frameMax := uint32(maxFrame) + ex := &exchange{ph: phCA, done: done, lastTCP: s.clock.Now()} label := "3.0" if major < 3 { - limit = maxPreauth20 - ph = phChallenge + frameMax = maxPreauth20 + ex.ph = phChallenge label = "2.0" } - wire := newFramed(conn, limit, done) + wire := newFramed(conn, frameMax) + go wire.read(done) s.mu.Lock() s.protocol = label s.rxSeq = 0 @@ -281,17 +301,9 @@ func (s *Station) session(ctx context.Context, conn net.Conn, major uint16) erro if err := writeFrame(wire, tcpProtoVer, versionWire(major, 0)); err != nil { return err } - now := s.clock.Now() - lastTCP := now + now := ex.lastTCP tcpRx := now.Add(15 * time.Second) udpRx := now.Add(60 * time.Second) - authed := false - - var udpCh chan []byte - var udpErr chan error - udpReady := false - var client tls.Certificate - var haveCert bool for { s.mu.Lock() @@ -299,14 +311,14 @@ func (s *Station) session(ctx context.Context, conn net.Conn, major uint16) erro silence, silenceOK := s.nextSilence() s.mu.Unlock() - next := lastTCP.Add(10 * time.Second) + next := ex.lastTCP.Add(10 * time.Second) if t := tcpRx; t.Before(next) { next = t } if t := udpRx; t.Before(next) { next = t } - if udpReady { + if ex.udpReady { if t := lastUDP.Add(15 * time.Second); t.Before(next) { next = t } @@ -322,9 +334,9 @@ func (s *Station) session(ctx context.Context, conn net.Conn, major uint16) erro var udpRecv <-chan []byte var udpFail <-chan error - if udpCh != nil { - udpRecv = udpCh - udpFail = udpErr + if ex.udpCh != nil { + udpRecv = ex.udpCh + udpFail = ex.udpErr } var reason string @@ -364,16 +376,16 @@ func (s *Station) session(ctx context.Context, conn net.Conn, major uint16) erro switch reason { case "tcp": tcpRx = now.Add(15 * time.Second) - limit := tcpRx - if udpRx.Before(limit) { - limit = udpRx + ex.limit = tcpRx + if udpRx.Before(ex.limit) { + ex.limit = udpRx } - err := s.onFrame(ctx, wire, major, tcpf, &ph, &authed, &client, &haveCert, done, &udpCh, &udpErr, &udpReady, &lastTCP, limit) + err := s.onFrame(ctx, wire, major, tcpf, ex) if err != nil { return err } case "udp": - if err := s.onUDP(wire, udpp, &ph, &udpRx, now, &lastTCP); err != nil { + if err := s.onUDP(wire, udpp, ex, &udpRx, now); err != nil { return err } default: @@ -381,19 +393,19 @@ func (s *Station) session(ctx context.Context, conn net.Conn, major uint16) erro s.onClock(now) s.mu.Unlock() if !now.Before(tcpRx) { - return s.retry("tcp receive timeout") + return s.retry("tcp receive timeout", nil) } if !now.Before(udpRx) { - return s.retry("udp receive timeout") + return s.retry("udp receive timeout", nil) } - if !now.Before(lastTCP.Add(10 * time.Second)) { + if !now.Before(ex.lastTCP.Add(10 * time.Second)) { if err := writeFrame(wire, tcpHeartbeat, nil); err != nil { return err } - lastTCP = s.clock.Now() + ex.lastTCP = s.clock.Now() } s.mu.Lock() - if udpReady && s.udp != nil && !s.clock.Now().Before(s.lastUDPSend.Add(15*time.Second)) { + if ex.udpReady && s.udp != nil && !s.clock.Now().Before(s.lastUDPSend.Add(15*time.Second)) { var err error if s.udpWait { err = s.sendRegistration() @@ -410,11 +422,15 @@ func (s *Station) session(ctx context.Context, conn net.Conn, major uint16) erro } } -func (s *Station) retry(reason string) error { +func (s *Station) retry(detail string, cause error) error { s.mu.Lock() - s.retryReason = reason + s.retryReason = detail s.mu.Unlock() - return errors.New(reason) + text := strings.ToLower(detail) + if cause == nil { + return errors.New(text) + } + return fmt.Errorf("%s: %w", text, cause) } func (s *Station) nextSilence() (time.Time, bool) { @@ -448,7 +464,7 @@ func (s *Station) onClock(now time.Time) { } } -func (s *Station) onFrame(ctx context.Context, wire *framed, major uint16, f frame, ph *phase, authed *bool, client *tls.Certificate, haveCert *bool, done chan struct{}, udpCh *chan []byte, udpErr *chan error, udpReady *bool, lastTCP *time.Time, limit time.Time) error { +func (s *Station) onFrame(ctx context.Context, wire *framed, major uint16, f frame, ex *exchange) error { switch f.typ { case tcpHeartbeat: return nil @@ -457,19 +473,20 @@ func (s *Station) onFrame(ctx context.Context, wire *framed, major uint16, f fra case tcpProtoDown: return s.onDowngrade(f.body, major) } - if !*authed && f.typ >= 100 { - return s.retry("Protocol error") + if !ex.authed && f.typ >= 100 { + return s.retry("Protocol error", nil) } if major < 3 { - return s.onV2(wire, f, ph, authed, done, udpCh, udpErr, udpReady, lastTCP) + return s.onV2(wire, f, ex) } - return s.onV3(ctx, wire, f, ph, authed, client, haveCert, done, udpCh, udpErr, udpReady, lastTCP, limit) + return s.onV3(ctx, wire, f, ex) } func (s *Station) onReflectorError(body []byte, major uint16) error { msg, _, err := parseString(body) if err != nil { - msg = "could not be read" + s.endFatalCause("could not be read", err, true) + return fmt.Errorf("could not be read: %w", err) } if major >= 3 && fallback20(msg) { return errFallback @@ -494,20 +511,20 @@ 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 { +func (s *Station) onV2(wire *framed, f frame, ex *exchange) error { switch f.typ { case tcpAuthOK: - *authed = true - *ph = phServer + ex.authed = true + ex.ph = phServer wire.setMax(maxFrame) return nil case tcpAuthChallenge: - return s.answerChallenge(wire, f.body, lastTCP) + return s.answerChallenge(wire, f.body, ex) case tcpServerInfo: - if *ph != phServer { - return s.retry("Protocol error") + if ex.ph != phServer { + return s.retry("Protocol error", nil) } - return s.finishV2(wire, f.body, done, udpCh, udpErr, udpReady, lastTCP) + return s.finishV2(wire, f.body, ex) case tcpTalkerStart: s.onTalker(f.body, true) return nil @@ -516,7 +533,7 @@ func (s *Station) onV2(wire *framed, f frame, ph *phase, authed *bool, done chan return nil case tcpNodeList, tcpNodeJoined, tcpNodeLeft, tcpRequestQSY, tcpStateEvent, tcpSignal, tcpTxStatus: if err := unpackIgnored(f); err != nil { - return s.retry("Protocol error") + return s.retry("Protocol error", err) } return nil default: @@ -524,9 +541,12 @@ func (s *Station) onV2(wire *framed, f frame, ph *phase, authed *bool, done chan } } -func (s *Station) answerChallenge(w io.Writer, body []byte, lastTCP *time.Time) error { +func (s *Station) answerChallenge(w io.Writer, body []byte, ex *exchange) error { chal, _, err := parseBlob(body) - if err != nil || len(chal) != 20 { + if err != nil { + return fmt.Errorf("bad challenge: %w", err) + } + if len(chal) != 20 { return errors.New("bad challenge") } mac := hmac.New(sha1.New, []byte(s.cfg.authKey)) @@ -536,204 +556,11 @@ func (s *Station) answerChallenge(w io.Writer, body []byte, lastTCP *time.Time) if err := writeFrame(w, tcpAuthResponse, out); err != nil { return err } - *lastTCP = s.clock.Now() + ex.lastTCP = s.clock.Now() return nil } -func (s *Station) onV3(ctx context.Context, wire *framed, f frame, ph *phase, authed *bool, client *tls.Certificate, haveCert *bool, done chan struct{}, udpCh *chan []byte, udpErr *chan error, udpReady *bool, lastTCP *time.Time, limit time.Time) error { - switch f.typ { - case tcpCAInfo: - if *ph != phCA { - return nil - } - return s.onCAInfo(wire, f.body, ph, lastTCP) - case tcpCABundle: - if *ph != phBundle { - return nil - } - return s.onCABundle(wire, f.body, ph, lastTCP) - case tcpStartEnc: - if *ph != phStart { - return nil - } - return s.onStartEncryption(ctx, wire, ph, client, haveCert, limit) - case tcpCsrReq: - return s.sendCSR(wire, lastTCP) - case tcpAuthChallenge: - return s.answerChallenge(wire, f.body, lastTCP) - case tcpAuthOK: - *authed = true - *ph = phServer - if csrNewerThanCert(s.cfg.certificateDir) { - if err := s.sendCSR(wire, lastTCP); err != nil { - return err - } - } - return nil - case tcpClientCert: - return s.saveCert(f.body) - case tcpServerInfo: - if !*authed { - return s.retry("Protocol error") - } - return s.finishV3Server(wire, f.body, ph, lastTCP) - case tcpStartUDPEnc: - if *ph != phUDP { - return nil - } - return s.finishV3UDP(wire.conn, done, udpCh, udpErr, udpReady, ph) - case tcpTalkerStart: - s.onTalker(f.body, true) - return nil - case tcpTalkerStop: - s.onTalker(f.body, false) - return nil - case tcpNodeList, tcpNodeJoined, tcpNodeLeft, tcpRequestQSY, tcpStateEvent, tcpSignal, tcpTxStatus: - if err := unpackIgnored(f); err != nil { - return s.retry("Protocol error") - } - return nil - default: - return nil - } -} - -func (s *Station) onCAInfo(w io.Writer, body []byte, ph *phase, lastTCP *time.Time) error { - _, digest, err := parseCAInfo(body) - if err != nil { - return s.retry("Protocol error") - } - if err := ensureCertDir(s.cfg.certificateDir); err != nil { - s.endFatal("Cannot create "+s.cfg.certificateDir, true) - return errStopped - } - sum, ok, err := caDigest(s.cfg.certificateDir) - if err != nil { - s.endFatal("Cannot create "+s.cfg.certificateDir, true) - return errStopped - } - if ok { - if !bytes.Equal(sum[:], digest) { - s.mu.Lock() - s.writeLog(s.clock.Now(), "CA digest differs") - s.mu.Unlock() - } - if err := writeFrame(w, tcpStartEncReq, nil); err != nil { - return err - } - *lastTCP = s.clock.Now() - *ph = phStart - return nil - } - if err := writeFrame(w, tcpCABundleReq, nil); err != nil { - return err - } - *lastTCP = s.clock.Now() - *ph = phBundle - return nil -} - -func parseCAInfo(b []byte) (uint16, []byte, error) { - size, rest, err := parseU16(b) - if err != nil { - return 0, nil, err - } - digest, rest, err := parseBlob(rest) - if err != nil { - return 0, nil, err - } - if len(digest) != sha256Len || len(rest) != 0 { - return 0, nil, io.ErrUnexpectedEOF - } - return size, digest, nil -} - -const sha256Len = 32 - -func (s *Station) onCABundle(w io.Writer, body []byte, ph *phase, lastTCP *time.Time) error { - caPEM, rest, err := parseString(body) - if err != nil { - return s.retry("CA signature rejected") - } - sig, rest, err := parseBlob(rest) - if err != nil { - return s.retry("CA signature rejected") - } - certPEM, rest, err := parseString(rest) - if err != nil || len(rest) != 0 { - return s.retry("CA signature rejected") - } - if err := verifyCABundle(caPEM, sig, certPEM); err != nil { - return s.retry("CA signature rejected") - } - if err := writeCABundle(s.cfg.certificateDir, caPEM); err != nil { - s.endFatal("Cannot create "+s.cfg.certificateDir, true) - return errStopped - } - if err := writeFrame(w, tcpStartEncReq, nil); err != nil { - return err - } - *lastTCP = s.clock.Now() - *ph = phStart - return nil -} - -func (s *Station) onStartEncryption(ctx context.Context, wire *framed, ph *phase, client *tls.Certificate, haveCert *bool, limit time.Time) error { - pair, present, fatal := prepareIdentity(s.cfg.certificateDir, s.cfg.callsign) - if fatal != "" { - s.endFatal(fatal, true) - return errStopped - } - *client = pair - *haveCert = present - caPEM, err := os.ReadFile(filepath.Join(s.cfg.certificateDir, caFileName)) - if err != nil { - s.endFatal("Cannot create "+s.cfg.certificateDir, true) - return errStopped - } - cfg, err := tlsConfig(s.cfg.host, caPEM, pair, present) - if err != nil { - return s.retry("TLS handshake failed") - } - if err := s.startTLS(ctx, wire, cfg, limit); err != nil { - if errors.Is(err, errHostMismatch) { - return s.retry("Reflector certificate does not match the host") - } - if ctx.Err() != nil { - return ctx.Err() - } - return s.retry("TLS handshake failed") - } - *ph = phIdent - return nil -} - -func (s *Station) sendCSR(w io.Writer, lastTCP *time.Time) error { - pem, err := readCSR(s.cfg.certificateDir) - if err != nil { - s.endFatal("Cannot create "+s.cfg.certificateDir, true) - return errStopped - } - if err := writeFrame(w, tcpCsr, appendString(nil, pem)); err != nil { - return err - } - *lastTCP = s.clock.Now() - return nil -} - -func (s *Station) saveCert(body []byte) error { - pem, _, err := parseString(body) - if err != nil || pem == "" { - return s.retry("Protocol error") - } - if err := writeClientCert(s.cfg.certificateDir, pem); err != nil { - s.endFatal("Cannot create "+s.cfg.certificateDir, true) - return errStopped - } - return errReconnect -} - -func (s *Station) finishV2(wire *framed, body []byte, done chan struct{}, udpCh *chan []byte, udpErr *chan error, udpReady *bool, lastTCP *time.Time) error { +func (s *Station) finishV2(wire *framed, body []byte, ex *exchange) error { id, enc, dec, err := s.acceptServer(body) if err != nil { return err @@ -759,7 +586,7 @@ func (s *Station) finishV2(wire *framed, body []byte, done chan struct{}, udpCh dec.Close() return err } - *lastTCP = s.clock.Now() + ex.lastTCP = s.clock.Now() udpConn, peer, err := dialAudio(wire.conn) if err != nil { enc.Close() @@ -768,7 +595,7 @@ func (s *Station) finishV2(wire *framed, body []byte, done chan struct{}, udpCh } ch := make(chan []byte, 32) errc := make(chan error, 1) - go readUDP(udpConn, peer, ch, errc, done) + go readUDP(udpConn, peer, ch, errc, ex.done) s.mu.Lock() s.clientID = id s.udpPeer = peer @@ -793,9 +620,9 @@ func (s *Station) finishV2(wire *framed, body []byte, done chan struct{}, udpCh s.writeLog(s.clock.Now(), "connected") s.publish() s.mu.Unlock() - *udpCh = ch - *udpErr = errc - *udpReady = true + ex.udpCh = ch + ex.udpErr = errc + ex.udpReady = true return nil } @@ -820,287 +647,6 @@ func (s *Station) acceptServer(body []byte) (uint16, *opus.Encoder, *opus.Decode return id, enc, dec, nil } -func (s *Station) finishV3Server(wire *framed, body []byte, ph *phase, lastTCP *time.Time) error { - id, enc, dec, err := s.acceptServer(body) - if err != nil { - return err - } - if enc == nil { - return errStopped - } - iv := make([]byte, 6) - key := make([]byte, 16) - if _, err := rand.Read(iv); err != nil || randFill(key) != nil { - enc.Close() - dec.Close() - return errors.New("udp key") - } - info, err := nodeInfoV3(iv, key) - if err != nil { - enc.Close() - dec.Close() - return err - } - if err := writeFrame(wire, tcpNodeInfo, info); err != nil { - enc.Close() - dec.Close() - return err - } - *lastTCP = s.clock.Now() - s.mu.Lock() - s.clientID = id - s.enc = enc - s.dec = dec - s.ivPrefix = iv - s.udpKey = key - s.secure = true - s.mu.Unlock() - *ph = phUDP - return nil -} - -func randFill(b []byte) error { - _, err := rand.Read(b) - return err -} - -func (s *Station) finishV3UDP(conn net.Conn, done chan struct{}, udpCh *chan []byte, udpErr *chan error, udpReady *bool, ph *phase) error { - udpConn, peer, err := dialAudio(conn) - if err != nil { - s.dropAudio() - return err - } - ch := make(chan []byte, 32) - errc := make(chan error, 1) - go readUDP(udpConn, peer, ch, errc, done) - s.mu.Lock() - s.udp = udpConn - s.udpPeer = peer - s.udpWait = true - s.txCounter = 1 - s.rxCounter = 0 - err = s.sendRegistration() - s.mu.Unlock() - if err != nil { - s.dropAudio() - udpConn.Close() - return err - } - *udpCh = ch - *udpErr = errc - *udpReady = true - *ph = phBeat - return nil -} - -func dialAudio(conn net.Conn) (*net.UDPConn, *net.UDPAddr, error) { - remote, ok := conn.RemoteAddr().(*net.TCPAddr) - if !ok { - return nil, nil, errors.New("tcp peer has no address") - } - local := &net.UDPAddr{Port: 0} - if tcpLocal, ok := conn.LocalAddr().(*net.TCPAddr); ok && tcpLocal.IP != nil { - local.IP = tcpLocal.IP - } - udpConn, err := net.ListenUDP("udp", local) - if err != nil { - return nil, nil, err - } - return udpConn, &net.UDPAddr{IP: append(net.IP(nil), remote.IP...), Port: remote.Port}, nil -} - -func (s *Station) dropAudio() { - s.mu.Lock() - defer s.mu.Unlock() - if s.enc != nil { - s.enc.Close() - s.enc = nil - } - if s.dec != nil { - s.dec.Close() - s.dec = nil - } -} - -func readUDP(conn *net.UDPConn, peer *net.UDPAddr, out chan<- []byte, errc chan<- error, done <-chan struct{}) { - buf := make([]byte, 2048) - for { - n, addr, err := conn.ReadFromUDP(buf) - if err != nil { - select { - case errc <- err: - case <-done: - } - return - } - if addr == nil || peer == nil || !addr.IP.Equal(peer.IP) || addr.Port != peer.Port { - continue - } - pkt := append([]byte(nil), buf[:n]...) - select { - case out <- pkt: - case <-done: - return - } - } -} - -func (s *Station) onUDP(w io.Writer, pkt []byte, ph *phase, udpRx *time.Time, now time.Time, lastTCP *time.Time) error { - s.mu.Lock() - defer s.mu.Unlock() - if s.secure { - plain, ok := s.openSecure(pkt) - if !ok { - return nil - } - *udpRx = now.Add(60 * time.Second) - if len(plain) < 2 { - return nil - } - typ := binary.BigEndian.Uint16(plain[:2]) - body := plain[2:] - if s.udpWait { - if typ != udpHeartbeat { - return nil - } - s.udpWait = false - s.mu.Unlock() - var tg [4]byte - binary.BigEndian.PutUint32(tg[:], uint32(s.cfg.talkGroup)) - err := writeFrame(w, tcpSelectTG, tg[:]) - s.mu.Lock() - if err != nil { - return err - } - *lastTCP = s.clock.Now() - s.state.Connection = "Connected" - s.up = true - s.outage = false - s.noteTransmitLocked() - s.writeLog(s.clock.Now(), "connected") - s.publish() - *ph = phRun - return nil - } - s.onPlain(typ, body, now) - return nil - } - if !s.acceptV2(pkt) { - return nil - } - *udpRx = now.Add(60 * time.Second) - typ := binary.BigEndian.Uint16(pkt[:2]) - s.onPlain(typ, pkt[6:], now) - return nil -} - -func (s *Station) openSecure(pkt []byte) ([]byte, bool) { - if len(pkt) < 12 || s.udpKey == nil { - return nil, false - } - counter := binary.BigEndian.Uint32(pkt[:4]) - if counter < s.rxCounter { - return nil, false - } - tag := pkt[4:12] - ct := pkt[12:] - plain, err := openDatagram(s.udpKey, s.ivPrefix, 0, counter, pkt[:4], tag, ct) - if err != nil { - return nil, false - } - s.rxCounter = counter + 1 - return plain, true -} - -func (s *Station) acceptV2(pkt []byte) bool { - if len(pkt) < 6 { - return false - } - if binary.BigEndian.Uint16(pkt[2:4]) != s.clientID { - return false - } - seq := binary.BigEndian.Uint16(pkt[4:6]) - diff := seq - s.rxSeq - if diff > 32767 { - return false - } - s.rxSeq = seq + 1 - return true -} - -func (s *Station) onPlain(typ uint16, body []byte, now time.Time) { - switch typ { - case udpAllFlushed: - s.flushWaiting = false - case udpAudio: - if s.dec == nil || s.transmitting() { - return - } - packet, _, err := parseBlob(body) - if err != nil { - return - } - pcm, err := s.dec.Decode(packet) - if err != nil { - return - } - s.play = append(s.play, pcm...) - s.roomBusy = true - s.audioHeard = true - s.lastAudio = now - s.publish() - } -} - -func (s *Station) sendRegistration() error { - return s.sendSecure(udpHeartbeat, nil, 0, true) -} - -func (s *Station) sendUDP(typ uint16, body []byte) error { - if s.udp == nil { - return errors.New("udp is down") - } - if s.secure { - return s.sendSecure(typ, body, s.txCounter, false) - } - buf := make([]byte, 6+len(body)) - binary.BigEndian.PutUint16(buf[0:2], typ) - binary.BigEndian.PutUint16(buf[2:4], s.clientID) - binary.BigEndian.PutUint16(buf[4:6], s.seq) - s.seq++ - copy(buf[6:], body) - if _, err := s.udp.WriteToUDP(buf, s.udpPeer); err != nil { - return err - } - s.lastUDPSend = s.clock.Now() - return nil -} - -func (s *Station) sendSecure(typ uint16, body []byte, counter uint32, reg bool) error { - plain := appendU16(nil, typ) - plain = append(plain, body...) - var aad []byte - id := s.clientID - if reg { - aad = registrationAAD(s.clientID) - counter = 0 - } else { - aad = counterAAD(counter) - } - wire, err := sealDatagram(s.udpKey, s.ivPrefix, id, counter, aad, plain) - if err != nil { - return err - } - if _, err := s.udp.WriteToUDP(wire, s.udpPeer); err != nil { - return err - } - s.lastUDPSend = s.clock.Now() - if !reg { - s.txCounter = counter + 1 - } - return nil -} - func (s *Station) stop(msg string) { s.endFatal(msg, false) } // Fail stops retries and shows message until the operator quits. diff --git a/station/session_test.go b/station/session_test.go index 8813afb..7d428f0 100644 --- a/station/session_test.go +++ b/station/session_test.go @@ -21,7 +21,10 @@ func TestLoginAnnouncesProtocolAndJoinsTheTalkGroup(t *testing.T) { } ctx, cancel := context.WithCancel(t.Context()) defer cancel() - go func() { _ = st.Run(ctx) }() + go func() { + // Run's error is observed through State, which is the station seam. + _ = st.Run(ctx) + }() got, err := st.Wait(ctx, func(s station.State) bool { return s.Connection == "Connecting" }) if err != nil { diff --git a/station/session_udp.go b/station/session_udp.go new file mode 100644 index 0000000..08cc062 --- /dev/null +++ b/station/session_udp.go @@ -0,0 +1,246 @@ +package station + +import ( + "encoding/binary" + "errors" + "io" + "net" + "time" +) + +func (s *Station) finishV3UDP(conn net.Conn, ex *exchange) 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, ex.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 + } + ex.udpCh = ch + ex.udpErr = errc + ex.udpReady = true + ex.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, ex *exchange, udpRx *time.Time, now 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 + } + ex.lastTCP = s.clock.Now() + s.state.Connection = "Connected" + s.up = true + s.outage = false + s.noteTransmitLocked() + s.writeLog(s.clock.Now(), "connected") + s.publish() + ex.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 +} diff --git a/station/session_v3.go b/station/session_v3.go new file mode 100644 index 0000000..e0baf3f --- /dev/null +++ b/station/session_v3.go @@ -0,0 +1,259 @@ +package station + +import ( + "bytes" + "context" + "crypto/rand" + "errors" + "fmt" + "io" + "os" + "path/filepath" +) + +func (s *Station) onV3(ctx context.Context, wire *framed, f frame, ex *exchange) error { + switch f.typ { + case tcpCAInfo: + if ex.ph != phCA { + return nil + } + return s.onCAInfo(wire, f.body, ex) + case tcpCABundle: + if ex.ph != phBundle { + return nil + } + return s.onCABundle(wire, f.body, ex) + case tcpStartEnc: + if ex.ph != phStart { + return nil + } + return s.onStartEncryption(ctx, wire, ex) + case tcpCsrReq: + return s.sendCSR(wire, ex) + case tcpAuthChallenge: + return s.answerChallenge(wire, f.body, ex) + case tcpAuthOK: + ex.authed = true + ex.ph = phServer + newer, err := csrNewerThanCert(s.cfg.certificateDir) + if err != nil { + s.endFatalCause("Cannot create "+s.cfg.certificateDir, err, true) + return errStopped + } + if newer { + if err := s.sendCSR(wire, ex); err != nil { + return err + } + } + return nil + case tcpClientCert: + return s.saveCert(f.body) + case tcpServerInfo: + if !ex.authed { + return s.retry("Protocol error", nil) + } + return s.finishV3Server(wire, f.body, ex) + case tcpStartUDPEnc: + if ex.ph != phUDP { + return nil + } + return s.finishV3UDP(wire.conn, ex) + case tcpTalkerStart: + s.onTalker(f.body, true) + return nil + case tcpTalkerStop: + s.onTalker(f.body, false) + return nil + case tcpNodeList, tcpNodeJoined, tcpNodeLeft, tcpRequestQSY, tcpStateEvent, tcpSignal, tcpTxStatus: + if err := unpackIgnored(f); err != nil { + return s.retry("Protocol error", err) + } + return nil + default: + return nil + } +} + +func (s *Station) onCAInfo(w io.Writer, body []byte, ex *exchange) error { + _, digest, err := parseCAInfo(body) + if err != nil { + return s.retry("Protocol error", err) + } + if err := ensureCertDir(s.cfg.certificateDir); err != nil { + s.endFatalCause("Cannot create "+s.cfg.certificateDir, err, true) + return errStopped + } + sum, ok, err := caDigest(s.cfg.certificateDir) + if err != nil { + s.endFatalCause("Cannot create "+s.cfg.certificateDir, err, 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 + } + ex.lastTCP = s.clock.Now() + ex.ph = phStart + return nil + } + if err := writeFrame(w, tcpCABundleReq, nil); err != nil { + return err + } + ex.lastTCP = s.clock.Now() + ex.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, ex *exchange) error { + caPEM, rest, err := parseString(body) + if err != nil { + return s.retry("CA signature rejected", err) + } + sig, rest, err := parseBlob(rest) + if err != nil { + return s.retry("CA signature rejected", err) + } + certPEM, rest, err := parseString(rest) + if err != nil { + return s.retry("CA signature rejected", err) + } + if len(rest) != 0 { + return s.retry("CA signature rejected", nil) + } + if err := verifyCABundle(caPEM, sig, certPEM); err != nil { + return s.retry("CA signature rejected", err) + } + if err := writeCABundle(s.cfg.certificateDir, caPEM); err != nil { + s.endFatalCause("Cannot create "+s.cfg.certificateDir, err, true) + return errStopped + } + if err := writeFrame(w, tcpStartEncReq, nil); err != nil { + return err + } + ex.lastTCP = s.clock.Now() + ex.ph = phStart + return nil +} + +func (s *Station) onStartEncryption(ctx context.Context, wire *framed, ex *exchange) error { + pair, present, fatal, cause := prepareIdentity(s.cfg.certificateDir, s.cfg.callsign) + if fatal != "" { + s.endFatalCause(fatal, cause, true) + return errStopped + } + ex.client = pair + ex.haveCert = present + caPath := filepath.Join(s.cfg.certificateDir, caFileName) + caPEM, err := os.ReadFile(caPath) + if err != nil { + s.endFatalCause("Cannot create "+caPath, err, true) + return errStopped + } + cfg, err := tlsConfig(s.cfg.host, caPEM, pair, present) + if err != nil { + return s.retry("TLS handshake failed", err) + } + if err := s.startTLS(ctx, wire, cfg, ex.limit); err != nil { + if errors.Is(err, errHostMismatch) { + return s.retry("Reflector certificate does not match the host", err) + } + if ctx.Err() != nil { + return ctx.Err() + } + return s.retry("TLS handshake failed", err) + } + ex.ph = phIdent + return nil +} + +func (s *Station) sendCSR(w io.Writer, ex *exchange) error { + csrPath := filepath.Join(s.cfg.certificateDir, csrFileName) + pem, err := readCSR(s.cfg.certificateDir) + if err != nil { + s.endFatalCause("Cannot create "+csrPath, err, true) + return errStopped + } + if err := writeFrame(w, tcpCsr, appendString(nil, pem)); err != nil { + return err + } + ex.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", err) + } + certPath := filepath.Join(s.cfg.certificateDir, certFileName) + if err := writeClientCert(s.cfg.certificateDir, pem); err != nil { + s.endFatalCause("Cannot create "+certPath, err, true) + return errStopped + } + return errReconnect +} +func (s *Station) finishV3Server(wire *framed, body []byte, ex *exchange) 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 { + enc.Close() + dec.Close() + return fmt.Errorf("udp key: %w", err) + } + if _, err := rand.Read(key); err != nil { + enc.Close() + dec.Close() + return fmt.Errorf("udp key: %w", err) + } + info, err := identityV3(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 + } + ex.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() + ex.ph = phUDP + return nil +} diff --git a/station/software.go b/station/software.go index ee70483..2ca5dbd 100644 --- a/station/software.go +++ b/station/software.go @@ -2,8 +2,8 @@ package station import ( "bytes" - jsonv2 "encoding/json/v2" + "golang.org/x/sys/unix" ) @@ -14,23 +14,23 @@ const ( SoftwareVersion = "1.0.0" ) -type nodeSoftware struct { +type softwareIdentity struct { SW string `json:"sw"` SWVer string `json:"swVer"` ProjVer string `json:"projVer"` - MachineArch string `json:"machineArch,omitempty"` + MachineArch string `json:"machineArch,omitzero"` } func softwareIdentityBody() ([]byte, error) { - raw, err := marshalNode(false) + raw, err := marshalIdentity(false) if err != nil { return nil, err } return appendString(nil, string(raw)), nil } -func nodeInfoV3(iv, key []byte) ([]byte, error) { - raw, err := marshalNode(true) +func identityV3(iv, key []byte) ([]byte, error) { + raw, err := marshalIdentity(true) if err != nil { return nil, err } @@ -39,21 +39,22 @@ func nodeInfoV3(iv, key []byte) ([]byte, error) { return appendString(body, string(raw)), nil } -func marshalNode(arch bool) ([]byte, error) { - info := nodeSoftware{ +func marshalIdentity(arch bool) ([]byte, error) { + info := softwareIdentity{ SW: SoftwareName, SWVer: SoftwareVersion, ProjVer: SoftwareVersion, } if arch { - if name, ok := machineArch(); ok { + if name, ok := MachineArch(); ok { info.MachineArch = name } } return jsonv2.Marshal(info) } -func machineArch() (string, bool) { +// MachineArch is the hardware name from uname, when it can be read. +func MachineArch() (string, bool) { var u unix.Utsname if err := unix.Uname(&u); err != nil { return "", false