konrad/funknetz
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity

Separate the window sentence from the wrapped error so a bad certificate file is not reported as a key mismatch.

Browse Source 
Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
konrad
2026-10-07 15:01:38 +02:00
co-authored by Cursor
parent a405048f23
commit e642b24be0
17 changed files with 842 additions and 742 deletions
Download Patch File Download Diff File Expand all files Collapse all files
station/behavior_test.go
+16 -4
View File
@@ -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)
station/cert.go
+51 -36
View File
@@ -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 {
station/cipher.go
+4 -56
View File
@@ -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)
}
station/config_test.go
+4 -1
View File
@@ -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()
station/lab_reflector_test.go
+4 -1
View File
@@ -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"
station/log.go
+9 -1
View File
@@ -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()
}
station/log_test.go
+8 -2
View File
@@ -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"))
station/protocol2_test.go
+12 -3
View File
@@ -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"}))
station/protocol3_test.go
+29 -64
View File
@@ -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)
station/protocol_select_test.go
+16 -4
View File
@@ -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 {
station/reflector_test.go
+1
View File
@@ -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
station/session.go
+105 -559
View File
@@ -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.
station/session_test.go
+4 -1
View File
@@ -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 {
station/session_udp.go
+246
View File
@@ -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
}
station/session_v3.go
+259
View File
@@ -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
}
station/software.go
+11 -10
View File
@@ -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