l2tp: implement traffic shaping plugin and RADIUS CoA/DM listener (spec-l2tp-8c-shaper)

New l2tp-shaper plugin subscribes to EventBus session lifecycle events
and programs TC qdiscs (TBF/HTB) on pppN interfaces via the existing
traffic.Backend abstraction. CoA/DM listener (RFC 5176) in the RADIUS
plugin receives rate-change and disconnect requests from RADIUS servers.

Components:
- l2tp/events: SessionUp and SessionRateChange typed EventBus events
- l2tp/reactor: emit SessionUp on ppp.EventSessionUp, store pppInterface
- radius/dict: CoA/DM packet codes (40-45), Error-Cause attribute (101)
- radius/packet: VerifyCoARequestAuth for RFC 5176 request validation
- l2tpshaper: plugin with config parsing, session state, TC application
- l2tpauthradius/coa: UDP listener with auth verification, session
  matching by Acct-Session-Id prefix or User-Name+NAS-Port, source
  address filtering from configured RADIUS servers
- traffic: export ParseRateBps for cross-package rate parsing
- rfc/short/rfc5176.md: RFC summary for Dynamic Authorization

Author: thomas-mangin

internal/component/l2tp/events/events.go

@@ -49,6 +49,39 @@ type SessionDownPayload struct {
 // plugin to release allocated IP addresses.
 var SessionDown = events.Register[*SessionDownPayload](Namespace, SessionDownEvent)
 
+// SessionUpEvent is the event type string for session-up notifications.
+const SessionUpEvent = "session-up"
+
+// SessionUpPayload carries session identity and the pppN interface name.
+// Emitted by the reactor when PPP LCP, authentication, and all enabled
+// NCPs complete successfully (ppp.EventSessionUp).
+type SessionUpPayload struct {
+	TunnelID  uint16
+	SessionID uint16
+	Interface string
+}
+
+// SessionUp is the typed handle for (l2tp, session-up). Consumed by
+// the shaper plugin to apply TC rules and by stats plugins.
+var SessionUp = events.Register[*SessionUpPayload](Namespace, SessionUpEvent)
+
+// SessionRateChangeEvent is the event type string for rate-change notifications.
+const SessionRateChangeEvent = "session-rate-change"
+
+// SessionRateChangePayload carries updated bandwidth for a session.
+// Emitted by the CoA handler in the RADIUS plugin when a RADIUS server
+// sends a CoA-Request with bandwidth attributes.
+type SessionRateChangePayload struct {
+	TunnelID     uint16
+	SessionID    uint16
+	DownloadRate uint64 // bits per second
+	UploadRate   uint64 // bits per second
+}
+
+// SessionRateChange is the typed handle for (l2tp, session-rate-change).
+// Consumed by the shaper plugin to update TC rules on the session's pppN.
+var SessionRateChange = events.Register[*SessionRateChangePayload](Namespace, SessionRateChangeEvent)
+
 // SessionIPAssignedEvent is the event type string for session-ip-assigned notifications.
 const SessionIPAssignedEvent = "session-ip-assigned"
 

internal/component/l2tp/events/events_test.go

@@ -28,3 +28,15 @@ func TestL2TP_RegisteredAsProducer(t *testing.T) {
 	require.True(t, slices.Contains(prods, ProtocolID),
 		"l2tp ProtocolID must appear in Producers()")
 }
+
+// VALIDATES: AC-1 prerequisite -- SessionUp typed handle exists.
+func TestSessionUpHandle_Registered(t *testing.T) {
+	require.Equal(t, "l2tp", SessionUp.Namespace())
+	require.Equal(t, SessionUpEvent, SessionUp.EventType())
+}
+
+// VALIDATES: SessionRateChange typed handle exists.
+func TestSessionRateChangeHandle_Registered(t *testing.T) {
+	require.Equal(t, "l2tp", SessionRateChange.Namespace())
+	require.Equal(t, SessionRateChangeEvent, SessionRateChange.EventType())
+}

internal/component/l2tp/reactor.go

@@ -845,6 +845,15 @@ func (r *L2TPReactor) handleKernelSuccess(ksucc kernelSetupSucceeded) {
 		ProxyLCPLastRecv:    ksucc.proxyLastRecvLCPConfReq,
 	}
 
+	ifaceName := fmt.Sprintf("ppp%d", ksucc.fds.unitNum)
+	r.tunnelsMu.Lock()
+	if tunnel, ok := r.tunnelsByLocalID[ksucc.localTID]; ok {
+		if sess := tunnel.lookupSession(ksucc.localSID); sess != nil {
+			sess.pppInterface = ifaceName
+		}
+	}
+	r.tunnelsMu.Unlock()
+
 	select {
 	case r.pppDriver.SessionsIn() <- start:
 	case <-r.stop:
@@ -879,7 +888,10 @@ func (r *L2TPReactor) handlePPPEvent(ev ppp.Event) {
 		tid, sid, reason = e.TunnelID, e.SessionID, e.Reason
 	case ppp.EventSessionRejected:
 		tid, sid, reason = e.TunnelID, e.SessionID, e.Reason
-	case ppp.EventLCPUp, ppp.EventLCPDown, ppp.EventSessionUp:
+	case ppp.EventLCPUp, ppp.EventLCPDown:
+		return
+	case ppp.EventSessionUp:
+		r.handleSessionUp(e)
 		return
 	}
 	if tid == 0 && sid == 0 {
@@ -977,6 +989,33 @@ func (r *L2TPReactor) handleSessionIPAssigned(ev ppp.EventSessionIPAssigned) {
 	}
 }
 
+// handleSessionUp emits the (l2tp, session-up) EventBus event when a
+// PPP session completes LCP, auth, and all NCPs. The shaper plugin
+// subscribes to this event to apply TC rules on the pppN interface.
+func (r *L2TPReactor) handleSessionUp(ev ppp.EventSessionUp) {
+	if r.eventBus == nil {
+		return
+	}
+	var ifaceName string
+	r.tunnelsMu.Lock()
+	if tunnel, ok := r.tunnelsByLocalID[ev.TunnelID]; ok {
+		if sess := tunnel.lookupSession(ev.SessionID); sess != nil {
+			ifaceName = sess.pppInterface
+		}
+	}
+	r.tunnelsMu.Unlock()
+	if ifaceName == "" {
+		return
+	}
+	if _, err := l2tpevents.SessionUp.Emit(r.eventBus, &l2tpevents.SessionUpPayload{
+		TunnelID:  ev.TunnelID,
+		SessionID: ev.SessionID,
+		Interface: ifaceName,
+	}); err != nil {
+		r.logger.Warn("l2tp: session-up emit failed", "error", err)
+	}
+}
+
 // handleKernelError processes a setup failure reported by the kernel
 // worker. Grabs tunnelsMu, looks up the session, and sends a CDN to
 // the peer if the session still exists.

internal/component/l2tp/session.go

@@ -115,6 +115,11 @@ type L2TPSession struct {
 	// (LNS side, true) or handleOCCN (LAC side, false). Used by the
 	// kernel worker to set L2TP_ATTR_LNS_MODE.
 	lnsMode bool
+
+	// pppInterface is the kernel pppN interface name (e.g. "ppp0").
+	// Set by handleKernelSuccess when the PPP session is started.
+	// Used by SessionUp EventBus event so the shaper can apply TC.
+	pppInterface string
 }
 
 // State returns the session's current FSM state.

internal/component/plugin/all/all.go

@@ -1,15 +1,26 @@
 // Design: docs/research/l2tpv2-ze-integration.md -- RADIUS attribute dictionary
+// Related: packet.go -- packet encode/decode consuming these codes
+// Related: client.go -- client transport using packet codes
+// Related: attr.go -- attribute encode/decode helpers
 
 package radius
 
-// RADIUS packet codes (RFC 2865 Section 3, RFC 2866 Section 3).
+// RADIUS packet codes (RFC 2865 Section 3, RFC 2866 Section 3, RFC 5176 Section 3).
 const (
 	CodeAccessRequest   = 1
 	CodeAccessAccept    = 2
 	CodeAccessReject    = 3
 	CodeAccountingReq   = 4
 	CodeAccountingResp  = 5
 	CodeAccessChallenge = 11
+
+	// RFC 5176 Section 3: Dynamic Authorization Extensions (CoA/DM).
+	CodeDisconnectRequest = 40
+	CodeDisconnectACK     = 41
+	CodeDisconnectNAK     = 42
+	CodeCoARequest        = 43
+	CodeCoAACK            = 44
+	CodeCoANAK            = 45
 )
 
 // RADIUS attribute type codes (RFC 2865 Section 5).
@@ -40,6 +51,7 @@ const (
 	AttrCHAPChallenge     = 60
 	AttrNASPortType       = 61
 	AttrFramedPool        = 88
+	AttrErrorCause        = 101 // RFC 5176 Section 3.6
 	AttrVendorSpecific    = 26
 )
 
@@ -77,6 +89,19 @@ const (
 	NASPortTypeVirtual = 5
 )
 
+// Error-Cause values (RFC 5176 Section 3.6).
+const (
+	ErrorCauseResidualSession      = 201
+	ErrorCauseInvalidEAPPacket     = 202
+	ErrorCauseUnsupportedAttribute = 401
+	ErrorCauseMissingAttribute     = 402
+	ErrorCauseNASIdentification    = 403
+	ErrorCauseInvalidRequest       = 404
+	ErrorCauseUnsupportedService   = 405
+	ErrorCauseUnsupportedExtension = 406
+	ErrorCauseSessionNotFound      = 503
+)
+
 // Wire constants.
 const (
 	HeaderLen        = 20 // Code(1) + ID(1) + Length(2) + Authenticator(16)

internal/component/radius/dict.go

@@ -172,8 +172,25 @@ func VerifyResponseAuth(response []byte, requestAuth [AuthenticatorLen]byte, sec
 	return subtle.ConstantTimeCompare(response[4:4+AuthenticatorLen], expected[:]) == 1
 }
 
+// VerifyCoARequestAuth checks the authenticator of a CoA-Request or
+// Disconnect-Request. RFC 5176 Section 3.5: same formula as
+// Accounting-Request (MD5 over Code+ID+Length+16-zero-octets+Attrs+Secret).
+// Uses constant-time comparison.
+func VerifyCoARequestAuth(data, secret []byte) bool {
+	if len(data) < MinPacketLen {
+		return false
+	}
+	pktLen := int(binary.BigEndian.Uint16(data[2:4]))
+	if pktLen < MinPacketLen || pktLen > len(data) {
+		return false
+	}
+	expected := AccountingRequestAuth(data, pktLen, secret)
+	return subtle.ConstantTimeCompare(data[4:4+AuthenticatorLen], expected[:]) == 1
+}
+
 // AccountingRequestAuth computes the authenticator for an Accounting-Request.
 // RFC 2866 Section 3: MD5(Code+ID+Length+16zero+Attributes+Secret).
+// RFC 5176 Section 3.5: same formula for CoA-Request and Disconnect-Request.
 func AccountingRequestAuth(buf []byte, length int, secret []byte) [AuthenticatorLen]byte {
 	h := md5.New() //nolint:gosec // RFC 2866 mandates MD5
 	h.Write(buf[:4])

internal/component/radius/packet.go

@@ -252,6 +252,105 @@ func TestAccountingRequestAuth(t *testing.T) {
 	}
 }
 
+// VALIDATES: AC-3/AC-4 -- CoA-Request (code 43) round-trips through encode/decode.
+func TestCoARequestRoundTrip(t *testing.T) {
+	pkt := &Packet{
+		Code:       CodeCoARequest,
+		Identifier: 10,
+		Attrs: []Attr{
+			{Type: AttrAcctSessionID, Value: AttrString("sess-001")},
+			{Type: AttrFilterID, Value: AttrString("10mbit")},
+		},
+	}
+
+	buf := make([]byte, MaxPacketLen)
+	n, err := pkt.EncodeTo(buf, 0)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	// Set proper CoA authenticator.
+	secret := []byte("coa-secret")
+	auth := AccountingRequestAuth(buf, n, secret)
+	copy(buf[4:4+AuthenticatorLen], auth[:])
+
+	decoded, err := Decode(buf[:n])
+	if err != nil {
+		t.Fatal(err)
+	}
+	if decoded.Code != CodeCoARequest {
+		t.Errorf("code: got %d, want %d", decoded.Code, CodeCoARequest)
+	}
+	sessID := decoded.FindAttr(AttrAcctSessionID)
+	if string(sessID) != "sess-001" {
+		t.Errorf("Acct-Session-Id: got %q, want %q", sessID, "sess-001")
+	}
+}
+
+// VALIDATES: AC-6/AC-7 -- Disconnect-Request (code 40) round-trips.
+func TestDisconnectRequestRoundTrip(t *testing.T) {
+	pkt := &Packet{
+		Code:       CodeDisconnectRequest,
+		Identifier: 20,
+		Attrs: []Attr{
+			{Type: AttrAcctSessionID, Value: AttrString("sess-002")},
+		},
+	}
+
+	buf := make([]byte, MaxPacketLen)
+	n, err := pkt.EncodeTo(buf, 0)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	decoded, err := Decode(buf[:n])
+	if err != nil {
+		t.Fatal(err)
+	}
+	if decoded.Code != CodeDisconnectRequest {
+		t.Errorf("code: got %d, want %d", decoded.Code, CodeDisconnectRequest)
+	}
+}
+
+// VALIDATES: AC-3 -- CoA request authenticator verification.
+func TestVerifyCoARequestAuth(t *testing.T) {
+	secret := []byte("coa-test-secret")
+	pkt := &Packet{
+		Code:       CodeCoARequest,
+		Identifier: 5,
+		Attrs: []Attr{
+			{Type: AttrAcctSessionID, Value: AttrString("sess-100")},
+		},
+	}
+
+	buf := make([]byte, MaxPacketLen)
+	n, err := pkt.EncodeTo(buf, 0)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	// Set correct authenticator.
+	auth := AccountingRequestAuth(buf, n, secret)
+	copy(buf[4:4+AuthenticatorLen], auth[:])
+
+	if !VerifyCoARequestAuth(buf[:n], secret) {
+		t.Error("valid CoA auth should verify")
+	}
+
+	// Corrupt one byte.
+	buf[4]++
+	if VerifyCoARequestAuth(buf[:n], secret) {
+		t.Error("corrupted CoA auth should fail verification")
+	}
+}
+
+// VALIDATES: AC-4 -- invalid authenticator on short packet.
+func TestVerifyCoARequestAuthTooShort(t *testing.T) {
+	if VerifyCoARequestAuth(make([]byte, 10), []byte("secret")) {
+		t.Error("short packet should fail verification")
+	}
+}
+
 func TestEncodeAtOffset(t *testing.T) {
 	pkt := &Packet{
 		Code:       CodeAccessRequest,

internal/component/radius/packet_test.go

@@ -86,15 +86,15 @@ func parseTrafficClass(name string, m map[string]any) (TrafficClass, error) {
 	tc := TrafficClass{Name: name}
 
 	if rateStr, ok := m["rate"].(string); ok {
-		rate, err := parseRateBps(rateStr)
+		rate, err := ParseRateBps(rateStr)
 		if err != nil {
 			return TrafficClass{}, fmt.Errorf("rate: %w", err)
 		}
 		tc.Rate = rate
 	}
 
 	if ceilStr, ok := m["ceil"].(string); ok {
-		ceil, err := parseRateBps(ceilStr)
+		ceil, err := ParseRateBps(ceilStr)
 		if err != nil {
 			return TrafficClass{}, fmt.Errorf("ceil: %w", err)
 		}
@@ -147,7 +147,7 @@ var rateSuffixes = []struct {
 	{"bps", 8},
 }
 
-func parseRateBps(v string) (uint64, error) {
+func ParseRateBps(v string) (uint64, error) {
 	for _, rs := range rateSuffixes {
 		if strings.HasSuffix(v, rs.suffix) {
 			numStr := v[:len(v)-len(rs.suffix)]

internal/component/traffic/config.go

@@ -67,19 +67,19 @@ func TestParseTrafficRateFormats(t *testing.T) {
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			got, err := parseRateBps(tt.rate)
+			got, err := ParseRateBps(tt.rate)
 			if err != nil {
-				t.Fatalf("parseRateBps(%q): %v", tt.rate, err)
+				t.Fatalf("ParseRateBps(%q): %v", tt.rate, err)
 			}
 			if got != tt.want {
-				t.Errorf("parseRateBps(%q) = %d, want %d", tt.rate, got, tt.want)
+				t.Errorf("ParseRateBps(%q) = %d, want %d", tt.rate, got, tt.want)
 			}
 		})
 	}
 }
 
 func TestParseTrafficInvalidRate(t *testing.T) {
-	_, err := parseRateBps("notarate")
+	_, err := ParseRateBps("notarate")
 	if err == nil {
 		t.Fatal("expected error for invalid rate")
 	}