[client] Add UI client event notifications (#3207)

This commit is contained in:
Viktor Liu 2025-02-20 11:00:02 +01:00 committed by GitHub
parent 87311074f1
commit 62a0c358f9
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
14 changed files with 1685 additions and 575 deletions

View File

@ -39,7 +39,6 @@ type peerStateDetailOutput struct {
TransferSent int64 `json:"transferSent" yaml:"transferSent"`
Latency time.Duration `json:"latency" yaml:"latency"`
RosenpassEnabled bool `json:"quantumResistance" yaml:"quantumResistance"`
Routes []string `json:"routes" yaml:"routes"`
Networks []string `json:"networks" yaml:"networks"`
}
@ -98,9 +97,9 @@ type statusOutputOverview struct {
FQDN string `json:"fqdn" yaml:"fqdn"`
RosenpassEnabled bool `json:"quantumResistance" yaml:"quantumResistance"`
RosenpassPermissive bool `json:"quantumResistancePermissive" yaml:"quantumResistancePermissive"`
Routes []string `json:"routes" yaml:"routes"`
Networks []string `json:"networks" yaml:"networks"`
NSServerGroups []nsServerGroupStateOutput `json:"dnsServers" yaml:"dnsServers"`
Events []systemEventOutput `json:"events" yaml:"events"`
}
var (
@ -284,9 +283,9 @@ func convertToStatusOutputOverview(resp *proto.StatusResponse) statusOutputOverv
FQDN: pbFullStatus.GetLocalPeerState().GetFqdn(),
RosenpassEnabled: pbFullStatus.GetLocalPeerState().GetRosenpassEnabled(),
RosenpassPermissive: pbFullStatus.GetLocalPeerState().GetRosenpassPermissive(),
Routes: pbFullStatus.GetLocalPeerState().GetNetworks(),
Networks: pbFullStatus.GetLocalPeerState().GetNetworks(),
NSServerGroups: mapNSGroups(pbFullStatus.GetDnsServers()),
Events: mapEvents(pbFullStatus.GetEvents()),
}
if anonymizeFlag {
@ -393,7 +392,6 @@ func mapPeers(peers []*proto.PeerState) peersStateOutput {
TransferSent: transferSent,
Latency: pbPeerState.GetLatency().AsDuration(),
RosenpassEnabled: pbPeerState.GetRosenpassEnabled(),
Routes: pbPeerState.GetNetworks(),
Networks: pbPeerState.GetNetworks(),
}
@ -559,7 +557,6 @@ func parseGeneralSummary(overview statusOutputOverview, showURL bool, showRelays
"NetBird IP: %s\n"+
"Interface type: %s\n"+
"Quantum resistance: %s\n"+
"Routes: %s\n"+
"Networks: %s\n"+
"Peers count: %s\n",
fmt.Sprintf("%s/%s%s", goos, goarch, goarm),
@ -574,7 +571,6 @@ func parseGeneralSummary(overview statusOutputOverview, showURL bool, showRelays
interfaceTypeString,
rosenpassEnabledStatus,
networks,
networks,
peersCountString,
)
return summary
@ -582,13 +578,17 @@ func parseGeneralSummary(overview statusOutputOverview, showURL bool, showRelays
func parseToFullDetailSummary(overview statusOutputOverview) string {
parsedPeersString := parsePeers(overview.Peers, overview.RosenpassEnabled, overview.RosenpassPermissive)
parsedEventsString := parseEvents(overview.Events)
summary := parseGeneralSummary(overview, true, true, true)
return fmt.Sprintf(
"Peers detail:"+
"%s\n"+
"Events:"+
"%s\n"+
"%s",
parsedPeersString,
parsedEventsString,
summary,
)
}
@ -657,7 +657,6 @@ func parsePeers(peers peersStateOutput, rosenpassEnabled, rosenpassPermissive bo
" Last WireGuard handshake: %s\n"+
" Transfer status (received/sent) %s/%s\n"+
" Quantum resistance: %s\n"+
" Routes: %s\n"+
" Networks: %s\n"+
" Latency: %s\n",
peerState.FQDN,
@ -676,7 +675,6 @@ func parsePeers(peers peersStateOutput, rosenpassEnabled, rosenpassPermissive bo
toIEC(peerState.TransferSent),
rosenpassEnabledStatus,
networks,
networks,
peerState.Latency.String(),
)
@ -825,14 +823,6 @@ func anonymizePeerDetail(a *anonymize.Anonymizer, peer *peerStateDetailOutput) {
for i, route := range peer.Networks {
peer.Networks[i] = a.AnonymizeRoute(route)
}
for i, route := range peer.Routes {
peer.Routes[i] = a.AnonymizeIPString(route)
}
for i, route := range peer.Routes {
peer.Routes[i] = a.AnonymizeRoute(route)
}
}
func anonymizeOverview(a *anonymize.Anonymizer, overview *statusOutputOverview) {
@ -870,9 +860,14 @@ func anonymizeOverview(a *anonymize.Anonymizer, overview *statusOutputOverview)
overview.Networks[i] = a.AnonymizeRoute(route)
}
for i, route := range overview.Routes {
overview.Routes[i] = a.AnonymizeRoute(route)
}
overview.FQDN = a.AnonymizeDomain(overview.FQDN)
for i, event := range overview.Events {
overview.Events[i].Message = a.AnonymizeString(event.Message)
overview.Events[i].UserMessage = a.AnonymizeString(event.UserMessage)
for k, v := range event.Metadata {
event.Metadata[k] = a.AnonymizeString(v)
}
}
}

View File

@ -0,0 +1,69 @@
package cmd
import (
"fmt"
"sort"
"strings"
"time"
"github.com/netbirdio/netbird/client/proto"
)
type systemEventOutput struct {
ID string `json:"id" yaml:"id"`
Severity string `json:"severity" yaml:"severity"`
Category string `json:"category" yaml:"category"`
Message string `json:"message" yaml:"message"`
UserMessage string `json:"userMessage" yaml:"userMessage"`
Timestamp time.Time `json:"timestamp" yaml:"timestamp"`
Metadata map[string]string `json:"metadata" yaml:"metadata"`
}
func mapEvents(protoEvents []*proto.SystemEvent) []systemEventOutput {
events := make([]systemEventOutput, len(protoEvents))
for i, event := range protoEvents {
events[i] = systemEventOutput{
ID: event.GetId(),
Severity: event.GetSeverity().String(),
Category: event.GetCategory().String(),
Message: event.GetMessage(),
UserMessage: event.GetUserMessage(),
Timestamp: event.GetTimestamp().AsTime(),
Metadata: event.GetMetadata(),
}
}
return events
}
func parseEvents(events []systemEventOutput) string {
if len(events) == 0 {
return " No events recorded"
}
var eventsString strings.Builder
for _, event := range events {
timeStr := timeAgo(event.Timestamp)
metadataStr := ""
if len(event.Metadata) > 0 {
pairs := make([]string, 0, len(event.Metadata))
for k, v := range event.Metadata {
pairs = append(pairs, fmt.Sprintf("%s: %s", k, v))
}
sort.Strings(pairs)
metadataStr = fmt.Sprintf("\n Metadata: %s", strings.Join(pairs, ", "))
}
eventsString.WriteString(fmt.Sprintf("\n [%s] %s (%s)"+
"\n Message: %s"+
"\n Time: %s%s",
event.Severity,
event.Category,
event.ID,
event.Message,
timeStr,
metadataStr,
))
}
return eventsString.String()
}

View File

@ -146,9 +146,6 @@ var overview = statusOutputOverview{
LastWireguardHandshake: time.Date(2001, 1, 1, 1, 1, 2, 0, time.UTC),
TransferReceived: 200,
TransferSent: 100,
Routes: []string{
"10.1.0.0/24",
},
Networks: []string{
"10.1.0.0/24",
},
@ -176,6 +173,7 @@ var overview = statusOutputOverview{
},
},
},
Events: []systemEventOutput{},
CliVersion: version.NetbirdVersion(),
DaemonVersion: "0.14.1",
ManagementState: managementStateOutput{
@ -230,9 +228,6 @@ var overview = statusOutputOverview{
Error: "timeout",
},
},
Routes: []string{
"10.10.0.0/24",
},
Networks: []string{
"10.10.0.0/24",
},
@ -299,9 +294,6 @@ func TestParsingToJSON(t *testing.T) {
"transferSent": 100,
"latency": 10000000,
"quantumResistance": false,
"routes": [
"10.1.0.0/24"
],
"networks": [
"10.1.0.0/24"
]
@ -327,7 +319,6 @@ func TestParsingToJSON(t *testing.T) {
"transferSent": 1000,
"latency": 10000000,
"quantumResistance": false,
"routes": null,
"networks": null
}
]
@ -366,9 +357,6 @@ func TestParsingToJSON(t *testing.T) {
"fqdn": "some-localhost.awesome-domain.com",
"quantumResistance": false,
"quantumResistancePermissive": false,
"routes": [
"10.10.0.0/24"
],
"networks": [
"10.10.0.0/24"
],
@ -393,7 +381,8 @@ func TestParsingToJSON(t *testing.T) {
"enabled": false,
"error": "timeout"
}
]
],
"events": []
}`
// @formatter:on
@ -429,8 +418,6 @@ func TestParsingToYAML(t *testing.T) {
transferSent: 100
latency: 10ms
quantumResistance: false
routes:
- 10.1.0.0/24
networks:
- 10.1.0.0/24
- fqdn: peer-2.awesome-domain.com
@ -451,7 +438,6 @@ func TestParsingToYAML(t *testing.T) {
transferSent: 1000
latency: 10ms
quantumResistance: false
routes: []
networks: []
cliVersion: development
daemonVersion: 0.14.1
@ -479,8 +465,6 @@ usesKernelInterface: true
fqdn: some-localhost.awesome-domain.com
quantumResistance: false
quantumResistancePermissive: false
routes:
- 10.10.0.0/24
networks:
- 10.10.0.0/24
dnsServers:
@ -497,6 +481,7 @@ dnsServers:
- example.net
enabled: false
error: timeout
events: []
`
assert.Equal(t, expectedYAML, yaml)
@ -526,7 +511,6 @@ func TestParsingToDetail(t *testing.T) {
Last WireGuard handshake: %s
Transfer status (received/sent) 200 B/100 B
Quantum resistance: false
Routes: 10.1.0.0/24
Networks: 10.1.0.0/24
Latency: 10ms
@ -543,10 +527,10 @@ func TestParsingToDetail(t *testing.T) {
Last WireGuard handshake: %s
Transfer status (received/sent) 2.0 KiB/1000 B
Quantum resistance: false
Routes: -
Networks: -
Latency: 10ms
Events: No events recorded
OS: %s/%s
Daemon version: 0.14.1
CLI version: %s
@ -562,7 +546,6 @@ FQDN: some-localhost.awesome-domain.com
NetBird IP: 192.168.178.100/16
Interface type: Kernel
Quantum resistance: false
Routes: 10.10.0.0/24
Networks: 10.10.0.0/24
Peers count: 2/2 Connected
`, lastConnectionUpdate1, lastHandshake1, lastConnectionUpdate2, lastHandshake2, runtime.GOOS, runtime.GOARCH, overview.CliVersion)
@ -584,7 +567,6 @@ FQDN: some-localhost.awesome-domain.com
NetBird IP: 192.168.178.100/16
Interface type: Kernel
Quantum resistance: false
Routes: 10.10.0.0/24
Networks: 10.10.0.0/24
Peers count: 2/2 Connected
`

View File

@ -68,6 +68,8 @@ type ConfigInput struct {
DisableFirewall *bool
BlockLANAccess *bool
DisableNotifications *bool
}
// Config Configuration type
@ -93,6 +95,8 @@ type Config struct {
BlockLANAccess bool
DisableNotifications bool
// SSHKey is a private SSH key in a PEM format
SSHKey string
@ -469,6 +473,16 @@ func (config *Config) apply(input ConfigInput) (updated bool, err error) {
updated = true
}
if input.DisableNotifications != nil && *input.DisableNotifications != config.DisableNotifications {
if *input.DisableNotifications {
log.Infof("disabling notifications")
} else {
log.Infof("enabling notifications")
}
config.DisableNotifications = *input.DisableNotifications
updated = true
}
if input.ClientCertKeyPath != "" {
config.ClientCertKeyPath = input.ClientCertKeyPath
updated = true

View File

@ -19,6 +19,7 @@ import (
log "github.com/sirupsen/logrus"
"github.com/netbirdio/netbird/client/internal/peer"
"github.com/netbirdio/netbird/client/proto"
)
const (
@ -230,6 +231,14 @@ func (u *upstreamResolverBase) probeAvailability() {
// didn't find a working upstream server, let's disable and try later
if !success {
u.disable(errors.ErrorOrNil())
u.statusRecorder.PublishEvent(
proto.SystemEvent_WARNING,
proto.SystemEvent_DNS,
"All upstream servers failed",
"Unable to reach one or more DNS servers. This might affect your ability to connect to some services.",
map[string]string{"upstreams": strings.Join(u.upstreamServers, ", ")},
)
}
}

View File

@ -7,21 +7,31 @@ import (
"sync"
"time"
"github.com/google/uuid"
log "github.com/sirupsen/logrus"
"golang.org/x/exp/maps"
"google.golang.org/grpc/codes"
gstatus "google.golang.org/grpc/status"
"google.golang.org/protobuf/types/known/timestamppb"
"github.com/netbirdio/netbird/client/iface/configurer"
"github.com/netbirdio/netbird/client/internal/relay"
"github.com/netbirdio/netbird/client/proto"
"github.com/netbirdio/netbird/management/domain"
relayClient "github.com/netbirdio/netbird/relay/client"
)
const eventQueueSize = 10
type ResolvedDomainInfo struct {
Prefixes []netip.Prefix
ParentDomain domain.Domain
}
type EventListener interface {
OnEvent(event *proto.SystemEvent)
}
// State contains the latest state of a peer
type State struct {
Mux *sync.RWMutex
@ -157,6 +167,10 @@ type Status struct {
peerListChangedForNotification bool
relayMgr *relayClient.Manager
eventMux sync.RWMutex
eventStreams map[string]chan *proto.SystemEvent
eventQueue *EventQueue
}
// NewRecorder returns a new Status instance
@ -164,6 +178,8 @@ func NewRecorder(mgmAddress string) *Status {
return &Status{
peers: make(map[string]State),
changeNotify: make(map[string]chan struct{}),
eventStreams: make(map[string]chan *proto.SystemEvent),
eventQueue: NewEventQueue(eventQueueSize),
offlinePeers: make([]State, 0),
notifier: newNotifier(),
mgmAddress: mgmAddress,
@ -806,3 +822,112 @@ func (d *Status) notifyAddressChanged() {
func (d *Status) numOfPeers() int {
return len(d.peers) + len(d.offlinePeers)
}
// PublishEvent adds an event to the queue and distributes it to all subscribers
func (d *Status) PublishEvent(
severity proto.SystemEvent_Severity,
category proto.SystemEvent_Category,
msg string,
userMsg string,
metadata map[string]string,
) {
event := &proto.SystemEvent{
Id: uuid.New().String(),
Severity: severity,
Category: category,
Message: msg,
UserMessage: userMsg,
Metadata: metadata,
Timestamp: timestamppb.Now(),
}
d.eventMux.Lock()
defer d.eventMux.Unlock()
d.eventQueue.Add(event)
for _, stream := range d.eventStreams {
select {
case stream <- event:
default:
log.Debugf("event stream buffer full, skipping event: %v", event)
}
}
log.Debugf("event published: %v", event)
}
// SubscribeToEvents returns a new event subscription
func (d *Status) SubscribeToEvents() *EventSubscription {
d.eventMux.Lock()
defer d.eventMux.Unlock()
id := uuid.New().String()
stream := make(chan *proto.SystemEvent, 10)
d.eventStreams[id] = stream
return &EventSubscription{
id: id,
events: stream,
}
}
// UnsubscribeFromEvents removes an event subscription
func (d *Status) UnsubscribeFromEvents(sub *EventSubscription) {
if sub == nil {
return
}
d.eventMux.Lock()
defer d.eventMux.Unlock()
if stream, exists := d.eventStreams[sub.id]; exists {
close(stream)
delete(d.eventStreams, sub.id)
}
}
// GetEventHistory returns all events in the queue
func (d *Status) GetEventHistory() []*proto.SystemEvent {
return d.eventQueue.GetAll()
}
type EventQueue struct {
maxSize int
events []*proto.SystemEvent
mutex sync.RWMutex
}
func NewEventQueue(size int) *EventQueue {
return &EventQueue{
maxSize: size,
events: make([]*proto.SystemEvent, 0, size),
}
}
func (q *EventQueue) Add(event *proto.SystemEvent) {
q.mutex.Lock()
defer q.mutex.Unlock()
q.events = append(q.events, event)
if len(q.events) > q.maxSize {
q.events = q.events[len(q.events)-q.maxSize:]
}
}
func (q *EventQueue) GetAll() []*proto.SystemEvent {
q.mutex.RLock()
defer q.mutex.RUnlock()
return slices.Clone(q.events)
}
type EventSubscription struct {
id string
events chan *proto.SystemEvent
}
func (s *EventSubscription) Events() <-chan *proto.SystemEvent {
return s.events
}

View File

@ -19,6 +19,7 @@ import (
"github.com/netbirdio/netbird/client/internal/routemanager/dynamic"
"github.com/netbirdio/netbird/client/internal/routemanager/refcounter"
"github.com/netbirdio/netbird/client/internal/routemanager/static"
"github.com/netbirdio/netbird/client/proto"
"github.com/netbirdio/netbird/route"
)
@ -28,6 +29,15 @@ const (
handlerTypeStatic
)
type reason int
const (
reasonUnknown reason = iota
reasonRouteUpdate
reasonPeerUpdate
reasonShutdown
)
type routerPeerStatus struct {
connected bool
relayed bool
@ -255,7 +265,7 @@ func (c *clientNetwork) removeRouteFromWireGuardPeer() error {
return nil
}
func (c *clientNetwork) removeRouteFromPeerAndSystem() error {
func (c *clientNetwork) removeRouteFromPeerAndSystem(rsn reason) error {
if c.currentChosen == nil {
return nil
}
@ -269,17 +279,19 @@ func (c *clientNetwork) removeRouteFromPeerAndSystem() error {
merr = multierror.Append(merr, fmt.Errorf("remove route: %w", err))
}
c.disconnectEvent(rsn)
return nberrors.FormatErrorOrNil(merr)
}
func (c *clientNetwork) recalculateRouteAndUpdatePeerAndSystem() error {
func (c *clientNetwork) recalculateRouteAndUpdatePeerAndSystem(rsn reason) error {
routerPeerStatuses := c.getRouterPeerStatuses()
newChosenID := c.getBestRouteFromStatuses(routerPeerStatuses)
// If no route is chosen, remove the route from the peer and system
if newChosenID == "" {
if err := c.removeRouteFromPeerAndSystem(); err != nil {
if err := c.removeRouteFromPeerAndSystem(rsn); err != nil {
return fmt.Errorf("remove route for peer %s: %w", c.currentChosen.Peer, err)
}
@ -319,6 +331,58 @@ func (c *clientNetwork) recalculateRouteAndUpdatePeerAndSystem() error {
return nil
}
func (c *clientNetwork) disconnectEvent(rsn reason) {
var defaultRoute bool
for _, r := range c.routes {
if r.Network.Bits() == 0 {
defaultRoute = true
break
}
}
if !defaultRoute {
return
}
var severity proto.SystemEvent_Severity
var message string
var userMessage string
meta := make(map[string]string)
switch rsn {
case reasonShutdown:
severity = proto.SystemEvent_INFO
message = "Default route removed"
userMessage = "Exit node disconnected."
meta["network"] = c.handler.String()
case reasonRouteUpdate:
severity = proto.SystemEvent_INFO
message = "Default route updated due to configuration change"
meta["network"] = c.handler.String()
case reasonPeerUpdate:
severity = proto.SystemEvent_WARNING
message = "Default route disconnected due to peer unreachability"
userMessage = "Exit node connection lost. Your internet access might be affected."
if c.currentChosen != nil {
meta["peer"] = c.currentChosen.Peer
meta["network"] = c.handler.String()
}
default:
severity = proto.SystemEvent_ERROR
message = "Default route disconnected for unknown reason"
userMessage = "Exit node disconnected for unknown reasons."
meta["network"] = c.handler.String()
}
c.statusRecorder.PublishEvent(
severity,
proto.SystemEvent_NETWORK,
message,
userMessage,
meta,
)
}
func (c *clientNetwork) sendUpdateToClientNetworkWatcher(update routesUpdate) {
go func() {
c.routeUpdate <- update
@ -361,12 +425,12 @@ func (c *clientNetwork) peersStateAndUpdateWatcher() {
select {
case <-c.ctx.Done():
log.Debugf("Stopping watcher for network [%v]", c.handler)
if err := c.removeRouteFromPeerAndSystem(); err != nil {
if err := c.removeRouteFromPeerAndSystem(reasonShutdown); err != nil {
log.Errorf("Failed to remove routes for [%v]: %v", c.handler, err)
}
return
case <-c.peerStateUpdate:
err := c.recalculateRouteAndUpdatePeerAndSystem()
err := c.recalculateRouteAndUpdatePeerAndSystem(reasonPeerUpdate)
if err != nil {
log.Errorf("Failed to recalculate routes for network [%v]: %v", c.handler, err)
}
@ -385,7 +449,7 @@ func (c *clientNetwork) peersStateAndUpdateWatcher() {
if isTrueRouteUpdate {
log.Debug("Client network update contains different routes, recalculating routes")
err := c.recalculateRouteAndUpdatePeerAndSystem()
err := c.recalculateRouteAndUpdatePeerAndSystem(reasonRouteUpdate)
if err != nil {
log.Errorf("Failed to recalculate routes for network [%v]: %v", c.handler, err)
}

File diff suppressed because it is too large Load Diff

View File

@ -59,6 +59,10 @@ service DaemonService {
rpc SetNetworkMapPersistence(SetNetworkMapPersistenceRequest) returns (SetNetworkMapPersistenceResponse) {}
rpc TracePacket(TracePacketRequest) returns (TracePacketResponse) {}
rpc SubscribeEvents(SubscribeRequest) returns (stream SystemEvent) {}
rpc GetEvents(GetEventsRequest) returns (GetEventsResponse) {}
}
@ -116,6 +120,8 @@ message LoginRequest {
optional bool disable_firewall = 23;
optional bool block_lan_access = 24;
optional bool disable_notifications = 25;
}
message LoginResponse {
@ -181,6 +187,8 @@ message GetConfigResponse {
bool rosenpassEnabled = 11;
bool rosenpassPermissive = 12;
bool disable_notifications = 13;
}
// PeerState contains the latest state of a peer
@ -251,6 +259,8 @@ message FullStatus {
repeated PeerState peers = 4;
repeated RelayState relays = 5;
repeated NSGroupState dns_servers = 6;
repeated SystemEvent events = 7;
}
message ListNetworksRequest {
@ -391,3 +401,35 @@ message TracePacketResponse {
repeated TraceStage stages = 1;
bool final_disposition = 2;
}
message SubscribeRequest{}
message SystemEvent {
enum Severity {
INFO = 0;
WARNING = 1;
ERROR = 2;
CRITICAL = 3;
}
enum Category {
NETWORK = 0;
DNS = 1;
AUTHENTICATION = 2;
CONNECTIVITY = 3;
}
string id = 1;
Severity severity = 2;
Category category = 3;
string message = 4;
string userMessage = 5;
google.protobuf.Timestamp timestamp = 6;
map<string, string> metadata = 7;
}
message GetEventsRequest {}
message GetEventsResponse {
repeated SystemEvent events = 1;
}

View File

@ -52,6 +52,8 @@ type DaemonServiceClient interface {
// SetNetworkMapPersistence enables or disables network map persistence
SetNetworkMapPersistence(ctx context.Context, in *SetNetworkMapPersistenceRequest, opts ...grpc.CallOption) (*SetNetworkMapPersistenceResponse, error)
TracePacket(ctx context.Context, in *TracePacketRequest, opts ...grpc.CallOption) (*TracePacketResponse, error)
SubscribeEvents(ctx context.Context, in *SubscribeRequest, opts ...grpc.CallOption) (DaemonService_SubscribeEventsClient, error)
GetEvents(ctx context.Context, in *GetEventsRequest, opts ...grpc.CallOption) (*GetEventsResponse, error)
}
type daemonServiceClient struct {
@ -215,6 +217,47 @@ func (c *daemonServiceClient) TracePacket(ctx context.Context, in *TracePacketRe
return out, nil
}
func (c *daemonServiceClient) SubscribeEvents(ctx context.Context, in *SubscribeRequest, opts ...grpc.CallOption) (DaemonService_SubscribeEventsClient, error) {
stream, err := c.cc.NewStream(ctx, &DaemonService_ServiceDesc.Streams[0], "/daemon.DaemonService/SubscribeEvents", opts...)
if err != nil {
return nil, err
}
x := &daemonServiceSubscribeEventsClient{stream}
if err := x.ClientStream.SendMsg(in); err != nil {
return nil, err
}
if err := x.ClientStream.CloseSend(); err != nil {
return nil, err
}
return x, nil
}
type DaemonService_SubscribeEventsClient interface {
Recv() (*SystemEvent, error)
grpc.ClientStream
}
type daemonServiceSubscribeEventsClient struct {
grpc.ClientStream
}
func (x *daemonServiceSubscribeEventsClient) Recv() (*SystemEvent, error) {
m := new(SystemEvent)
if err := x.ClientStream.RecvMsg(m); err != nil {
return nil, err
}
return m, nil
}
func (c *daemonServiceClient) GetEvents(ctx context.Context, in *GetEventsRequest, opts ...grpc.CallOption) (*GetEventsResponse, error) {
out := new(GetEventsResponse)
err := c.cc.Invoke(ctx, "/daemon.DaemonService/GetEvents", in, out, opts...)
if err != nil {
return nil, err
}
return out, nil
}
// DaemonServiceServer is the server API for DaemonService service.
// All implementations must embed UnimplementedDaemonServiceServer
// for forward compatibility
@ -253,6 +296,8 @@ type DaemonServiceServer interface {
// SetNetworkMapPersistence enables or disables network map persistence
SetNetworkMapPersistence(context.Context, *SetNetworkMapPersistenceRequest) (*SetNetworkMapPersistenceResponse, error)
TracePacket(context.Context, *TracePacketRequest) (*TracePacketResponse, error)
SubscribeEvents(*SubscribeRequest, DaemonService_SubscribeEventsServer) error
GetEvents(context.Context, *GetEventsRequest) (*GetEventsResponse, error)
mustEmbedUnimplementedDaemonServiceServer()
}
@ -311,6 +356,12 @@ func (UnimplementedDaemonServiceServer) SetNetworkMapPersistence(context.Context
func (UnimplementedDaemonServiceServer) TracePacket(context.Context, *TracePacketRequest) (*TracePacketResponse, error) {
return nil, status.Errorf(codes.Unimplemented, "method TracePacket not implemented")
}
func (UnimplementedDaemonServiceServer) SubscribeEvents(*SubscribeRequest, DaemonService_SubscribeEventsServer) error {
return status.Errorf(codes.Unimplemented, "method SubscribeEvents not implemented")
}
func (UnimplementedDaemonServiceServer) GetEvents(context.Context, *GetEventsRequest) (*GetEventsResponse, error) {
return nil, status.Errorf(codes.Unimplemented, "method GetEvents not implemented")
}
func (UnimplementedDaemonServiceServer) mustEmbedUnimplementedDaemonServiceServer() {}
// UnsafeDaemonServiceServer may be embedded to opt out of forward compatibility for this service.
@ -630,6 +681,45 @@ func _DaemonService_TracePacket_Handler(srv interface{}, ctx context.Context, de
return interceptor(ctx, in, info, handler)
}
func _DaemonService_SubscribeEvents_Handler(srv interface{}, stream grpc.ServerStream) error {
m := new(SubscribeRequest)
if err := stream.RecvMsg(m); err != nil {
return err
}
return srv.(DaemonServiceServer).SubscribeEvents(m, &daemonServiceSubscribeEventsServer{stream})
}
type DaemonService_SubscribeEventsServer interface {
Send(*SystemEvent) error
grpc.ServerStream
}
type daemonServiceSubscribeEventsServer struct {
grpc.ServerStream
}
func (x *daemonServiceSubscribeEventsServer) Send(m *SystemEvent) error {
return x.ServerStream.SendMsg(m)
}
func _DaemonService_GetEvents_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
in := new(GetEventsRequest)
if err := dec(in); err != nil {
return nil, err
}
if interceptor == nil {
return srv.(DaemonServiceServer).GetEvents(ctx, in)
}
info := &grpc.UnaryServerInfo{
Server: srv,
FullMethod: "/daemon.DaemonService/GetEvents",
}
handler := func(ctx context.Context, req interface{}) (interface{}, error) {
return srv.(DaemonServiceServer).GetEvents(ctx, req.(*GetEventsRequest))
}
return interceptor(ctx, in, info, handler)
}
// DaemonService_ServiceDesc is the grpc.ServiceDesc for DaemonService service.
// It's only intended for direct use with grpc.RegisterService,
// and not to be introspected or modified (even as a copy)
@ -705,7 +795,17 @@ var DaemonService_ServiceDesc = grpc.ServiceDesc{
MethodName: "TracePacket",
Handler: _DaemonService_TracePacket_Handler,
},
{
MethodName: "GetEvents",
Handler: _DaemonService_GetEvents_Handler,
},
},
Streams: []grpc.StreamDesc{
{
StreamName: "SubscribeEvents",
Handler: _DaemonService_SubscribeEvents_Handler,
ServerStreams: true,
},
},
Streams: []grpc.StreamDesc{},
Metadata: "daemon.proto",
}

36
client/server/event.go Normal file
View File

@ -0,0 +1,36 @@
package server
import (
"context"
log "github.com/sirupsen/logrus"
"github.com/netbirdio/netbird/client/proto"
)
func (s *Server) SubscribeEvents(req *proto.SubscribeRequest, stream proto.DaemonService_SubscribeEventsServer) error {
subscription := s.statusRecorder.SubscribeToEvents()
defer func() {
s.statusRecorder.UnsubscribeFromEvents(subscription)
log.Debug("client unsubscribed from events")
}()
log.Debug("client subscribed to events")
for {
select {
case event := <-subscription.Events():
if err := stream.Send(event); err != nil {
log.Warnf("error sending event to %v: %v", req, err)
return err
}
case <-stream.Context().Done():
return nil
}
}
}
func (s *Server) GetEvents(context.Context, *proto.GetEventsRequest) (*proto.GetEventsResponse, error) {
events := s.statusRecorder.GetEventHistory()
return &proto.GetEventsResponse{Events: events}, nil
}

View File

@ -404,6 +404,11 @@ func (s *Server) Login(callerCtx context.Context, msg *proto.LoginRequest) (*pro
s.latestConfigInput.BlockLANAccess = msg.BlockLanAccess
}
if msg.DisableNotifications != nil {
inputConfig.DisableNotifications = msg.DisableNotifications
s.latestConfigInput.DisableNotifications = msg.DisableNotifications
}
s.mutex.Unlock()
if msg.OptionalPreSharedKey != nil {
@ -687,6 +692,7 @@ func (s *Server) Status(
fullStatus := s.statusRecorder.GetFullStatus()
pbFullStatus := toProtoFullStatus(fullStatus)
pbFullStatus.Events = s.statusRecorder.GetEventHistory()
statusResponse.FullStatus = pbFullStatus
}
@ -736,17 +742,18 @@ func (s *Server) GetConfig(_ context.Context, _ *proto.GetConfigRequest) (*proto
}
return &proto.GetConfigResponse{
ManagementUrl: managementURL,
ConfigFile: s.latestConfigInput.ConfigPath,
LogFile: s.logFile,
PreSharedKey: preSharedKey,
AdminURL: adminURL,
InterfaceName: s.config.WgIface,
WireguardPort: int64(s.config.WgPort),
DisableAutoConnect: s.config.DisableAutoConnect,
ServerSSHAllowed: *s.config.ServerSSHAllowed,
RosenpassEnabled: s.config.RosenpassEnabled,
RosenpassPermissive: s.config.RosenpassPermissive,
ManagementUrl: managementURL,
ConfigFile: s.latestConfigInput.ConfigPath,
LogFile: s.logFile,
PreSharedKey: preSharedKey,
AdminURL: adminURL,
InterfaceName: s.config.WgIface,
WireguardPort: int64(s.config.WgPort),
DisableAutoConnect: s.config.DisableAutoConnect,
ServerSSHAllowed: *s.config.ServerSSHAllowed,
RosenpassEnabled: s.config.RosenpassEnabled,
RosenpassPermissive: s.config.RosenpassPermissive,
DisableNotifications: s.config.DisableNotifications,
}, nil
}
func (s *Server) onSessionExpire() {

View File

@ -34,6 +34,7 @@ import (
"github.com/netbirdio/netbird/client/internal"
"github.com/netbirdio/netbird/client/proto"
"github.com/netbirdio/netbird/client/system"
"github.com/netbirdio/netbird/client/ui/event"
"github.com/netbirdio/netbird/util"
"github.com/netbirdio/netbird/version"
)
@ -161,6 +162,7 @@ type serviceClient struct {
mAllowSSH *systray.MenuItem
mAutoConnect *systray.MenuItem
mEnableRosenpass *systray.MenuItem
mNotifications *systray.MenuItem
mAdvancedSettings *systray.MenuItem
// application with main windows.
@ -196,6 +198,8 @@ type serviceClient struct {
isUpdateIconActive bool
showRoutes bool
wRoutes fyne.Window
eventManager *event.Manager
}
// newServiceClient instance constructor
@ -429,6 +433,7 @@ func (s *serviceClient) menuUpClick() error {
log.Errorf("up service: %v", err)
return err
}
return nil
}
@ -570,6 +575,7 @@ func (s *serviceClient) onTrayReady() {
s.mAllowSSH = s.mSettings.AddSubMenuItemCheckbox("Allow SSH", "Allow SSH connections", false)
s.mAutoConnect = s.mSettings.AddSubMenuItemCheckbox("Connect on Startup", "Connect automatically when the service starts", false)
s.mEnableRosenpass = s.mSettings.AddSubMenuItemCheckbox("Enable Quantum-Resistance", "Enable post-quantum security via Rosenpass", false)
s.mNotifications = s.mSettings.AddSubMenuItemCheckbox("Notifications", "Enable notifications", true)
s.mAdvancedSettings = s.mSettings.AddSubMenuItem("Advanced Settings", "Advanced settings of the application")
s.loadSettings()
@ -606,6 +612,10 @@ func (s *serviceClient) onTrayReady() {
}
}()
s.eventManager = event.NewManager(s.app, s.addr)
s.eventManager.SetNotificationsEnabled(s.mNotifications.Checked())
go s.eventManager.Start(s.ctx)
go func() {
var err error
for {
@ -680,7 +690,20 @@ func (s *serviceClient) onTrayReady() {
defer s.mRoutes.Enable()
s.runSelfCommand("networks", "true")
}()
case <-s.mNotifications.ClickedCh:
if s.mNotifications.Checked() {
s.mNotifications.Uncheck()
} else {
s.mNotifications.Check()
}
if s.eventManager != nil {
s.eventManager.SetNotificationsEnabled(s.mNotifications.Checked())
}
if err := s.updateConfig(); err != nil {
log.Errorf("failed to update config: %v", err)
}
}
if err != nil {
log.Errorf("process connection: %v", err)
}
@ -780,8 +803,20 @@ func (s *serviceClient) getSrvConfig() {
if !cfg.RosenpassEnabled {
s.sRosenpassPermissive.Disable()
}
}
if s.mNotifications == nil {
return
}
if cfg.DisableNotifications {
s.mNotifications.Uncheck()
} else {
s.mNotifications.Check()
}
if s.eventManager != nil {
s.eventManager.SetNotificationsEnabled(s.mNotifications.Checked())
}
}
func (s *serviceClient) onUpdateAvailable() {
@ -846,6 +881,15 @@ func (s *serviceClient) loadSettings() {
} else {
s.mEnableRosenpass.Uncheck()
}
if cfg.DisableNotifications {
s.mNotifications.Uncheck()
} else {
s.mNotifications.Check()
}
if s.eventManager != nil {
s.eventManager.SetNotificationsEnabled(s.mNotifications.Checked())
}
}
// updateConfig updates the configuration parameters
@ -854,12 +898,14 @@ func (s *serviceClient) updateConfig() error {
disableAutoStart := !s.mAutoConnect.Checked()
sshAllowed := s.mAllowSSH.Checked()
rosenpassEnabled := s.mEnableRosenpass.Checked()
notificationsDisabled := !s.mNotifications.Checked()
loginRequest := proto.LoginRequest{
IsLinuxDesktopClient: runtime.GOOS == "linux",
ServerSSHAllowed: &sshAllowed,
RosenpassEnabled: &rosenpassEnabled,
DisableAutoConnect: &disableAutoStart,
DisableNotifications: &notificationsDisabled,
}
if err := s.restartClient(&loginRequest); err != nil {

151
client/ui/event/event.go Normal file
View File

@ -0,0 +1,151 @@
package event
import (
"context"
"fmt"
"strings"
"sync"
"time"
"fyne.io/fyne/v2"
"github.com/cenkalti/backoff/v4"
log "github.com/sirupsen/logrus"
"google.golang.org/grpc"
"google.golang.org/grpc/credentials/insecure"
"github.com/netbirdio/netbird/client/proto"
"github.com/netbirdio/netbird/client/system"
)
type Manager struct {
app fyne.App
addr string
mu sync.Mutex
ctx context.Context
cancel context.CancelFunc
enabled bool
}
func NewManager(app fyne.App, addr string) *Manager {
return &Manager{
app: app,
addr: addr,
}
}
func (e *Manager) Start(ctx context.Context) {
e.mu.Lock()
e.ctx, e.cancel = context.WithCancel(ctx)
e.mu.Unlock()
expBackOff := backoff.WithContext(&backoff.ExponentialBackOff{
InitialInterval: time.Second,
RandomizationFactor: backoff.DefaultRandomizationFactor,
Multiplier: backoff.DefaultMultiplier,
MaxInterval: 10 * time.Second,
MaxElapsedTime: 0,
Stop: backoff.Stop,
Clock: backoff.SystemClock,
}, ctx)
if err := backoff.Retry(e.streamEvents, expBackOff); err != nil {
log.Errorf("event stream ended: %v", err)
}
}
func (e *Manager) streamEvents() error {
e.mu.Lock()
ctx := e.ctx
e.mu.Unlock()
client, err := getClient(e.addr)
if err != nil {
return fmt.Errorf("create client: %w", err)
}
stream, err := client.SubscribeEvents(ctx, &proto.SubscribeRequest{})
if err != nil {
return fmt.Errorf("failed to subscribe to events: %w", err)
}
log.Info("subscribed to daemon events")
defer func() {
log.Info("unsubscribed from daemon events")
}()
for {
event, err := stream.Recv()
if err != nil {
return fmt.Errorf("error receiving event: %w", err)
}
e.handleEvent(event)
}
}
func (e *Manager) Stop() {
e.mu.Lock()
defer e.mu.Unlock()
if e.cancel != nil {
e.cancel()
}
}
func (e *Manager) SetNotificationsEnabled(enabled bool) {
e.mu.Lock()
defer e.mu.Unlock()
e.enabled = enabled
}
func (e *Manager) handleEvent(event *proto.SystemEvent) {
e.mu.Lock()
enabled := e.enabled
e.mu.Unlock()
if !enabled {
return
}
title := e.getEventTitle(event)
e.app.SendNotification(fyne.NewNotification(title, event.UserMessage))
}
func (e *Manager) getEventTitle(event *proto.SystemEvent) string {
var prefix string
switch event.Severity {
case proto.SystemEvent_ERROR, proto.SystemEvent_CRITICAL:
prefix = "Error"
case proto.SystemEvent_WARNING:
prefix = "Warning"
default:
prefix = "Info"
}
var category string
switch event.Category {
case proto.SystemEvent_DNS:
category = "DNS"
case proto.SystemEvent_NETWORK:
category = "Network"
case proto.SystemEvent_AUTHENTICATION:
category = "Authentication"
case proto.SystemEvent_CONNECTIVITY:
category = "Connectivity"
default:
category = "System"
}
return fmt.Sprintf("%s: %s", prefix, category)
}
func getClient(addr string) (proto.DaemonServiceClient, error) {
conn, err := grpc.NewClient(
strings.TrimPrefix(addr, "tcp://"),
grpc.WithTransportCredentials(insecure.NewCredentials()),
grpc.WithUserAgent(system.GetDesktopUIUserAgent()),
)
if err != nil {
return nil, err
}
return proto.NewDaemonServiceClient(conn), nil
}