diff --git a/internal/agent/agent.go b/internal/agent/agent.go index d9b505c..a09e2ce 100644 --- a/internal/agent/agent.go +++ b/internal/agent/agent.go @@ -26,6 +26,7 @@ type Agent struct { httpServer *http.Server start func() error stop func() error + reload func() } // AgentReq is the configuration used to construct a new Agent. @@ -39,6 +40,10 @@ type AgentReq struct { // Stop is called by Agent.Stop after the HTTP server has been shut down. Stop func() error + + // Reload is an optional function called when SIGHUP is received. It + // should trigger a non-blocking reload of dynamic configuration. + Reload func() } // New constructs an Agent from the provided AgentReq. If HTTP is enabled in @@ -53,6 +58,7 @@ func New(req *AgentReq) (*Agent, error) { logger: req.Logger.Named(log.ComponentNameAgent), start: req.Start, stop: req.Stop, + reload: req.Reload, } if req.HTTPConfig != nil && req.HTTPConfig.Enabled != nil && *req.HTTPConfig.Enabled { @@ -134,7 +140,10 @@ func (a *Agent) WaitForSignal() { switch sig { case syscall.SIGHUP: - a.logger.Info("SIGHUP received, configuration reload not yet implemented") + a.logger.Info("SIGHUP received, reloading agent") + if a.reload != nil { + a.reload() + } default: a.logger.Info("shutting down") if err := a.Stop(); err != nil { diff --git a/internal/agent/client.go b/internal/agent/client.go index 84aeb98..1a3c973 100644 --- a/internal/agent/client.go +++ b/internal/agent/client.go @@ -44,5 +44,6 @@ func NewClientAgent(cfg *config.ClientAgentConfig) (*Agent, error) { HTTPConfig: cfg.HTTP, Start: cl.Start, Stop: cl.Stop, + Reload: cl.Reload, }) } diff --git a/internal/client/client.go b/internal/client/client.go index 32a86a6..b05682e 100644 --- a/internal/client/client.go +++ b/internal/client/client.go @@ -44,11 +44,23 @@ type Client struct { networkManager *network.Manager // networks tracks the networks that this Smuggle client is aware of and - // should configure on the host. - networks []*types.Network + // should configure on the host. This is used to compare against updates + // from the store to determine when networks are added or removed. All + // access to this slice must be synchronized using the networksLock mutex. + networks []*types.Network + networksLock sync.RWMutex + + subnetWatchers map[string]*subnetWatcher + subnetWatchersLock sync.Mutex + + // subnetHeartbeaters tracks the currently running subnet heartbeaters. It + // is keyed by subnet ID and all access must be synchronized using the + // subnetHeartbeatersLock mutex. + subnetHeartbeaters map[string]*heartbeater + subnetHeartbeatersLock sync.Mutex // - subnets []*types.Subnet + reloadCh chan struct{} // shutdownCh is used to signal to all client processes that the agent is // shutting down. All long-running processes should monitor this channel and @@ -73,31 +85,44 @@ func New(req *ClientReq) (*Client, error) { } return &Client{ - cfg: req.Config, - logger: req.Logger.Named(log.ComponentNameClient), - networks: []*types.Network{}, - store: req.Store, - cniStore: req.CNIStore, - networkManager: netManager, - shutdownCh: make(chan struct{}), + cfg: req.Config, + logger: req.Logger.Named(log.ComponentNameClient), + networks: []*types.Network{}, + store: req.Store, + cniStore: req.CNIStore, + networkManager: netManager, + reloadCh: make(chan struct{}, 1), + shutdownCh: make(chan struct{}), + subnetWatchers: make(map[string]*subnetWatcher), + subnetHeartbeaters: make(map[string]*heartbeater), }, nil } +// Reload signals the client to immediately re-read the network list from the +// store. If a reload is already pending, the signal is dropped to avoid +// queuing up redundant work. +func (c *Client) Reload() { + select { + case c.reloadCh <- struct{}{}: + default: + // If the channel is full, do not block. This means a reload is already + // pending, so we do not need to send another signal. + } +} + func (c *Client) Start() error { if err := c.generateID(); err != nil { return fmt.Errorf("failed to get client ID: %w", err) } - if err := c.Init(); err != nil { - return fmt.Errorf("failed to initialize client: %w", err) - } - - if err := c.startSubnetUpdateHandler(); err != nil { - return fmt.Errorf("failed to start remote subnet handler: %w", err) - } + // Perform the initial network setup synchronously so that all local subnets + // and remote subnet routes are in place before Start returns. The + // background monitor handles periodic refreshes and SIGHUP reloads after + // this point. + c.triggerNetworksRead() - c.startHeartbeaters() + go c.monitorNetworks() return nil } @@ -131,107 +156,7 @@ func (c *Client) Stop() error { return nil } -func (c *Client) Init() error { - - // Read all network configurations from the store that we are able to see - // and therefore should configure on this host. - listResp, err := c.store.ListNetworks(nil) - if err != nil { - return fmt.Errorf("failed to get network configs: %w", err) - } - - if len(listResp.Networks) == 0 { - return errors.New("no networks configurations found") - } - - for _, networkConfig := range listResp.Networks { - - // Validate the network configuration. - if err := networkConfig.Validate(); err != nil { - return fmt.Errorf("invalid network: %w", err) - } - - c.networks = append(c.networks, networkConfig) - - clientSubnetResp, err := c.store.GetSubnet(&types.StoreGetSubnetReq{ - ID: c.id.Load().(string), - NetworkName: networkConfig.Name, - }) - if err != nil { - return fmt.Errorf("failed to get client subnet config: %w", err) - } - - // Perform the canonicalization, so we have all fields set correctly - // set. It would be possible to write this back to the data store, but - // seeing as this happens on the client, if more than one started at the - // same time, they would all race to write it back. - networkConfig.Canonicalize() - - subnet := clientSubnetResp.Subnet - - if subnet == nil { - - // - subnetListReq := types.StoreListSubnetsReq{Network: networkConfig.Name} - - subnetListResp, err := c.store.ListSubnets(&subnetListReq) - if err != nil { - return fmt.Errorf("failed to list existing client subnets: %w", err) - } - - subnet, err = c.networkManager.GenerateIPv4Subnet(c.getID(), networkConfig, subnetListResp.Subnets) - if err != nil { - return fmt.Errorf("failed to generate IPv4 subnet: %w", err) - } - } - - c.logger.Info("initializing local host subnet", networkConfig.LoggingPairs()...) - - if err := c.initSubnet(networkConfig, subnet); err != nil { - return fmt.Errorf("failed to initialize subnet: %w", err) - } - - c.subnets = append(c.subnets, subnet) - - if networkConfig.IPMasq != nil && *networkConfig.IPMasq { - if err := c.networkManager.Firewall.SetupMasqRules(networkConfig, subnet); err != nil { - return fmt.Errorf("failed to set up firewall masquerade rules: %w", err) - } - } - - if err := c.networkManager.Firewall.SetupForwardRules(networkConfig); err != nil { - return fmt.Errorf("failed to set up firewall forward rules: %w", err) - } - - c.logger.Info("successfully initialized local host subnet", subnet.LoggingPairs()...) - } - - if err := c.networkManager.Firewall.EnsureIsolation(c.networks); err != nil { - return fmt.Errorf("failed to ensure network isolation: %w", err) - } - - return nil -} - -func (c *Client) initSubnet(netCfg *types.Network, cfg *types.Subnet) error { - - providerResp, err := c.networkManager.SetLocal(&types.NetworkProviderSetReq{Client: cfg}) - if err != nil { - return fmt.Errorf("failed to set up local subnet: %w", err) - } - - if _, err := c.store.SetSubnet(&types.StoreSetSubnetReq{ - Subnet: providerResp.Network, - }); err != nil { - return fmt.Errorf("failed to store client subnet: %w", err) - } - - if err := c.cniStore.Set(types.GenerateCNIConfig(netCfg, cfg)); err != nil { - return fmt.Errorf("failed to write CNI config: %w", err) - } - - return nil -} +func (c *Client) shutdownGroupDecrement() { c.shutdownGroup.Done() } // generateID attempts to read the client ID from disk. If the file does not exist, // it generates a new UUID, saves it to disk, and returns it. diff --git a/internal/client/heartbeat.go b/internal/client/heartbeat.go index 902b6dd..0a043be 100644 --- a/internal/client/heartbeat.go +++ b/internal/client/heartbeat.go @@ -1,6 +1,7 @@ package client import ( + "sync" "time" "go.uber.org/zap" @@ -8,15 +9,35 @@ import ( "github.com/rasorp/smuggle/internal/types" ) -func (c *Client) startHeartbeaters() { - for _, subnet := range c.subnets { - go c.startSubnetHeartbeat(subnet) - } +// heartbeater is responsible for periodically updating the expiration time +// of a subnet in the store to indicate that the client is still active and +// using that subnet. +type heartbeater struct { + logger *zap.Logger + store types.Store + subnet *types.Subnet + + // shutdownCh is the client shutdown channel used to indicate that the agent + // is shutting down and all long-running processes should exit. This is a + // coarse-grained signal. + shutdownCh chan struct{} + + // stopCh is used to signal to this specific heartbeater that it should + // stop. This allows for more fine-grained control, such as when a subnet is + // removed. + stopCh chan struct{} + + // stopWGFn is a callback function that should be called when the + // heartbeater has fully stopped. This allows for proper coordination of + // shutdown processes across the client. + stopWGFn func() + + // stopOnce ensures that the stop process is only initiated once, preventing + // potential issues from multiple stop signals. + stopOnce sync.Once } -func (c *Client) startSubnetHeartbeat(subnet *types.Subnet) { - c.shutdownGroup.Add(1) - defer c.shutdownGroup.Done() +func (h *heartbeater) start() { // Calculate the heartbeat interval as half of the TTL to ensure we update // before expiration. This provides a safety margin. @@ -25,29 +46,25 @@ func (c *Client) startSubnetHeartbeat(subnet *types.Subnet) { ticker := time.NewTicker(heartbeatInterval) defer ticker.Stop() - // We may log more than one message, so caputre the pairs here to avoid - // multiple calls to the function and slice allocations. All fields are - // static. - logPairs := subnet.LoggingPairs() - - c.logger.Info("starting subnet heartbeat", - append(logPairs, zap.String("interval", heartbeatInterval.String()))..., + h.logger.Info("starting subnet heartbeat", + append(h.subnet.LoggingPairs(), zap.String("interval", heartbeatInterval.String()))..., ) + // This is a small codebase currently and we known this cannot be nil. In + // the future, if this code is refactored or reused in other contexts, we + // may want to add some additional safety checks or validation. + defer h.stopWGFn() + for { select { case <-ticker.C: // Create a copy of the subnet config to update the expiration time // without modifying the original reference. Then write this update // back to the store. - subnetCopy := subnet.Copy() + subnetCopy := h.subnet.Copy() subnetCopy.Expiration = time.Now().Add(types.DefaultSubnetTTL) - c.logger.Debug("updating subnet expiration", - append(logPairs, zap.Time("expiration", subnetCopy.Expiration))..., - ) - - _, err := c.store.SetSubnet(&types.StoreSetSubnetReq{Subnet: subnetCopy}) + _, err := h.store.SetSubnet(&types.StoreSetSubnetReq{Subnet: subnetCopy}) // Adjust the ticker interval based on success or failure. On // success, we maintain the regular interval. On failure, we shorten @@ -60,19 +77,28 @@ func (c *Client) startSubnetHeartbeat(subnet *types.Subnet) { case nil: ticker.Reset(types.DefaultSubnetTTL / 3) - c.logger.Info("successfully updated subnet expiration", - append(logPairs, zap.Time("expiration", subnetCopy.Expiration))..., + h.subnet = subnetCopy + + h.logger.Debug("updated subnet expiration", + zap.String("network", subnetCopy.NetworkName), + zap.Time("new_expiration", subnetCopy.Expiration), ) default: ticker.Reset(10 * time.Second) - c.logger.Error("failed to update subnet expiration", - append(logPairs, zap.Error(err))..., + h.logger.Error("failed to update subnet expiration", + zap.String("network", subnetCopy.NetworkName), + zap.Error(err), ) } - case <-c.shutdownCh: - c.logger.Info("shutting down subnet heartbeat", logPairs...) + case <-h.shutdownCh: + h.logger.Info("shutting down subnet heartbeat", zap.String("network", h.subnet.NetworkName)) + return + case <-h.stopCh: + h.logger.Info("stopping subnet heartbeat", zap.String("network", h.subnet.NetworkName)) return } } } + +func (h *heartbeater) stop() { h.stopOnce.Do(func() { close(h.stopCh) }) } diff --git a/internal/client/network.go b/internal/client/network.go new file mode 100644 index 0000000..b2af226 --- /dev/null +++ b/internal/client/network.go @@ -0,0 +1,425 @@ +package client + +import ( + "fmt" + "time" + + "go.uber.org/zap" + + "github.com/rasorp/smuggle/internal/types" +) + +func (c *Client) monitorNetworks() { + c.shutdownGroup.Add(1) + defer c.shutdownGroup.Done() + + // The ticker duration is currently hardcoded to 10 minutes, which is long + // enough to avoid excessive load on the store, but short enough to ensure + // that network changes are picked up in a timely manner. + // + // If operators require more immediate updates, they can issue a SIGHUP to + // the agent process to trigger an immediate refresh of the network list. + ticker := time.NewTicker(10 * time.Minute) + defer ticker.Stop() + + for { + select { + case <-ticker.C: + c.triggerNetworksRead() + case <-c.reloadCh: + c.triggerNetworksRead() + ticker.Reset(10 * time.Minute) + case <-c.shutdownCh: + c.logger.Info("shutting down network monitor") + return + } + } +} + +// triggerNetworksRead fetches the list of networks from the store and +// processes any additions, updates, or deletions. +func (c *Client) triggerNetworksRead() { + listResp, err := c.store.ListNetworks(nil) + if err != nil { + c.logger.Error("failed to list networks", zap.Error(err)) + } else { + c.logger.Info("successfully listed networks", zap.Int("network_num", len(listResp.Networks))) + c.handleNetworkMonitorTrigger(listResp.Networks) + c.logger.Info("network reload complete") + } +} + +func (c *Client) handleNetworkMonitorTrigger(nets []*types.Network) { + + // Lock the client's network list for the duration of this function, as we + // will be modifying it based on the list of networks from the store. While + // it's a coarse lock, it simplifies the logic and is acceptable given that + // network updates are expected to be infrequent and the operations + // performed here are not expected to be long-running. + c.networksLock.Lock() + defer c.networksLock.Unlock() + + // Build maps for easier comparison. + currentNetworks := make(map[string]*types.Network, len(c.networks)) + for _, network := range c.networks { + currentNetworks[network.Name] = network + } + + // Iterate through the array of networks from the store. Validate and + // canonicalize each one here, so we don't have to worry about it later. + // + // Invalid networks are logged and skipped. This protects us from a bad + // update performing an incorrect reconfiguration of a network on the host. + newNetworks := make(map[string]*types.Network, len(nets)) + + for _, network := range nets { + if err := network.Validate(); err != nil { + c.logger.Error("invalid network configuration", + zap.String("network_name", network.Name), + zap.Error(err), + ) + } else { + network.Canonicalize() + newNetworks[network.Name] = network + } + } + + var networkAdditions, networkUpdates, networksToDelete []*types.Network + + // Identify additions and updates. + for name, newNet := range newNetworks { + if currentNet, exists := currentNetworks[name]; !exists { + networkAdditions = append(networkAdditions, newNet) + } else if !currentNet.Equals(newNet) { + networkUpdates = append(networkUpdates, newNet) + } + } + + // Identify deletions. + deleteSet := make(map[string]struct{}, len(networksToDelete)) + for name, network := range currentNetworks { + if _, exists := newNetworks[name]; !exists { + networksToDelete = append(networksToDelete, network) + deleteSet[name] = struct{}{} + } + } + + // Track successful operations for firewall updates. + successfulAdditions := make([]*types.Network, 0, len(networkAdditions)) + successfulDeletions := make([]*types.Network, 0, len(networksToDelete)) + + // Handle network deletions first to free up resources. + for _, network := range networksToDelete { + c.deleteNetwork(network) + successfulDeletions = append(successfulDeletions, network) + } + + // Remove deleted networks from slice using map for O(1) lookup. + if len(deleteSet) > 0 { + updatedNetworks := make([]*types.Network, 0, len(c.networks)-len(deleteSet)) + for _, network := range c.networks { + if _, isDeleted := deleteSet[network.Name]; !isDeleted { + updatedNetworks = append(updatedNetworks, network) + } + } + c.networks = updatedNetworks + + // Delete isolation rules for removed networks. + if err := c.networkManager.Firewall.DeleteIsolation( + c.networks, successfulDeletions, + ); err != nil { + c.logger.Error("failed to delete network isolation", zap.Error(err)) + } + } + + // Handle network updates (delete then re-add). + for _, network := range networkUpdates { + c.logger.Info("detected network configuration change, updating", network.LoggingPairs()...) + // Remove old version from the slice. + for i, n := range c.networks { + if n.Name == network.Name { + c.networks = append(c.networks[:i], c.networks[i+1:]...) + break + } + } + c.deleteNetwork(network) + c.addNetwork(network) + } + + // Handle network additions. + for _, network := range networkAdditions { + c.addNetwork(network) + successfulAdditions = append(successfulAdditions, network) + } + + // Update isolation if we had any additions or updates. + if len(successfulAdditions) > 0 || len(networkUpdates) > 0 { + if err := c.networkManager.Firewall.CreateIsolation(c.networks); err != nil { + c.logger.Error("failed to ensure network isolation", zap.Error(err)) + } + } +} + +// addNetwork sets up a new network on the client, including subnet +// initialization, firewall configuration, and starting the heartbeating +// and subnet watching processes. +func (c *Client) addNetwork(network *types.Network) { + + subnet, err := c.setupNetwork(network) + if err != nil { + c.logger.Error("failed to setup new network", + append(network.LoggingPairs(), zap.Error(err))..., + ) + } else { + + // Explicitly set up routing for all remote subnets that already exist in + // the store. This is necessary to avoid a race on startup where the watch + // goroutine delivers initial state asynchronously, but traffic may be + // attempted before it has had a chance to run. + c.syncRemoteSubnets(network, subnet) + + // Setup and start the heartbeater for the new subnet. Once started, + // store it in the client's map of subnet heartbeaters, so it can be + // managed when needed. + hb := &heartbeater{ + logger: c.logger, + store: c.store, + subnet: subnet, + shutdownCh: c.shutdownCh, + stopCh: make(chan struct{}), + stopWGFn: c.shutdownGroupDecrement, + } + + c.shutdownGroup.Add(1) + go hb.start() + + c.subnetHeartbeatersLock.Lock() + c.subnetHeartbeaters[network.Name] = hb + c.subnetHeartbeatersLock.Unlock() + + // Setup and start the subnet watcher for the new network. Once started, + // store it in the client's map of subnet watchers, so it can be managed + // when needed. + sw := &subnetWatcher{ + logger: c.logger, + store: c.store, + cID: c.getID(), + network: network, + networkManager: c.networkManager, + localSubnet: subnet, + shutdownCh: c.shutdownCh, + stopCh: make(chan struct{}), + stopWGFn: c.shutdownGroupDecrement, + } + + c.shutdownGroup.Add(1) + sw.start() + + c.subnetWatchersLock.Lock() + c.subnetWatchers[network.Name] = sw + c.subnetWatchersLock.Unlock() + + c.networks = append(c.networks, network) + + c.logger.Info("successfully set up new network", network.LoggingPairs()...) + } +} + +// syncRemoteSubnets lists all currently known remote subnets for the given +// network and synchronously installs their routes on this host. This is called +// during network initialisation to avoid a startup race: the watch goroutine +// delivers existing state asynchronously, so without this explicit sync there +// is a window between Start returning and routes being in place. +func (c *Client) syncRemoteSubnets(network *types.Network, localSubnet *types.Subnet) { + + listResp, err := c.store.ListSubnets(&types.StoreListSubnetsReq{Network: network.Name}) + if err != nil { + c.logger.Error("failed to list remote subnets for initial sync", + append(network.LoggingPairs(), zap.Error(err))..., + ) + return + } + + var localSubnets []*types.Subnet + if localSubnet != nil { + localSubnets = []*types.Subnet{localSubnet} + } + + clientID := c.getID() + + for _, subnet := range listResp.Subnets { + if subnet.ClientID == clientID || subnet.Expired { + continue + } + + c.logger.Debug("setting up remote subnet during network initialization", + subnet.LoggingPairs()...) + + _, err := c.networkManager.SetRemote(&types.NetworkProviderSetRemoteReq{ + Subnet: subnet, + LocalSubnets: localSubnets, + }) + if err != nil { + c.logger.Error("failed to set up remote subnet during initialization", + append(subnet.LoggingPairs(), zap.Error(err))..., + ) + } + } +} + +// setupNetwork performs the complete setup for a new network, including subnet +// initialization and firewall configuration. +func (c *Client) setupNetwork(networkConfig *types.Network) (*types.Subnet, error) { + + clientSubnetResp, err := c.store.GetSubnet(&types.StoreGetSubnetReq{ + ID: c.id.Load().(string), + NetworkName: networkConfig.Name, + }) + if err != nil { + return nil, fmt.Errorf("failed to get client subnet config: %w", err) + } + + subnet := clientSubnetResp.Subnet + + // If there is no record of a subnet for this client, we need to generate + // a new one. Otherwise, we can use the existing one. + if subnet == nil { + + // We need to list all existing subnets for this network, so we can + // avoid IP conflicts when generating a new subnet. + subnetListResp, err := c.store.ListSubnets( + &types.StoreListSubnetsReq{ + Network: networkConfig.Name, + }, + ) + if err != nil { + return nil, fmt.Errorf("failed to list existing client subnets: %w", err) + } + + subnet, err = c.networkManager.GenerateIPv4Subnet(c.getID(), networkConfig, subnetListResp.Subnets) + if err != nil { + return nil, fmt.Errorf("failed to generate IPv4 subnet: %w", err) + } + } + + c.logger.Info("initializing local host subnet", networkConfig.LoggingPairs()...) + + if err := c.initNetworkSubnet(networkConfig, subnet); err != nil { + return nil, fmt.Errorf("failed to initialize subnet: %w", err) + } + + if networkConfig.IPMasq != nil && *networkConfig.IPMasq { + if err := c.networkManager.Firewall.CreateMasqRules(networkConfig, subnet); err != nil { + return nil, fmt.Errorf("failed to set up firewall masquerade rules: %w", err) + } + } + + if err := c.networkManager.Firewall.CreateForwardRules(networkConfig); err != nil { + return nil, fmt.Errorf("failed to set up firewall forward rules: %w", err) + } + + c.logger.Info("successfully initialized local host subnet", subnet.LoggingPairs()...) + return subnet, nil +} + +func (c *Client) initNetworkSubnet(netCfg *types.Network, cfg *types.Subnet) error { + + providerResp, err := c.networkManager.SetLocal(&types.NetworkProviderSetReq{Client: cfg}) + if err != nil { + return fmt.Errorf("failed to set up local subnet: %w", err) + } + + if _, err := c.store.SetSubnet(&types.StoreSetSubnetReq{ + Subnet: providerResp.Network, + }); err != nil { + return fmt.Errorf("failed to store client subnet: %w", err) + } + + if err := c.cniStore.Set(types.GenerateCNIConfig(netCfg, cfg)); err != nil { + return fmt.Errorf("failed to write CNI config: %w", err) + } + + return nil +} + +func (c *Client) deleteNetwork(network *types.Network) { + + networkPairs := network.LoggingPairs() + + clientSubnetResp, err := c.store.GetSubnet(&types.StoreGetSubnetReq{ + ID: c.id.Load().(string), + NetworkName: network.Name, + }) + + switch err { + case nil: + + // If there is no subnet for this network, we can skip the resource + // cleanup as there won't be any resources to clean up. This can happen + // if the network was added but the subnet initialization failed, so we + // never got to the point of creating the subnet record in the store. + if clientSubnetResp.Subnet == nil { + c.logger.Warn("no subnet found for network, skipping resource cleanup", networkPairs...) + break + } + subnetPairs := clientSubnetResp.Subnet.LoggingPairs() + c.logger.Debug("deleting local host subnet", subnetPairs...) + + // Remove firewall rules first, so that any traffic to/from the + // subnet is blocked before we remove the network interfaces. + if network.IPMasq != nil && *network.IPMasq { + if err := c.networkManager.Firewall.DeleteMasqRules( + network, + clientSubnetResp.Subnet, + ); err != nil { + c.logger.Error("failed to remove firewall masquerade rules", + append(subnetPairs, zap.Error(err))..., + ) + } + } + + if err := c.networkManager.Firewall.DeleteForwardRules(network); err != nil { + c.logger.Error("failed to remove firewall forward rules", + append(subnetPairs, zap.Error(err))..., + ) + } + + // Delete the local subnet networking. + if _, err := c.networkManager.DeleteLocal(&types.NetworkProviderDeleteLocalReq{ + Subnet: clientSubnetResp.Subnet, + }); err != nil { + c.logger.Error("failed to delete local subnet networking", + append(subnetPairs, zap.Error(err))..., + ) + } + + default: + c.logger.Error("failed to get subnet", append(networkPairs, zap.Error(err))...) + } + + // The CNI configuration file deletion does not require having the subnet + // information, so we can try and delete it. The function does not return an + // error if the file does not exist, so this is safe to do and tidy from an + // operational perspective. + if err := c.cniStore.Delete(network.Name); err != nil { + c.logger.Error("failed to delete CNI config", append(networkPairs, zap.Error(err))...) + } + + // Even if we fail to get the subnet, we still attempt to stop and clear the + // heartbeater and watcher, as they may still be running. + c.subnetHeartbeatersLock.Lock() + if hb, exists := c.subnetHeartbeaters[network.Name]; exists { + hb.stop() + delete(c.subnetHeartbeaters, network.Name) + } + c.subnetHeartbeatersLock.Unlock() + + c.subnetWatchersLock.Lock() + if watcher, exists := c.subnetWatchers[network.Name]; exists { + watcher.stop() + delete(c.subnetWatchers, network.Name) + } + c.subnetWatchersLock.Unlock() + + c.logger.Info("successfully deleted network", network.LoggingPairs()...) +} diff --git a/internal/client/subnet.go b/internal/client/subnet.go index 60e16e9..7b3cb91 100644 --- a/internal/client/subnet.go +++ b/internal/client/subnet.go @@ -2,106 +2,150 @@ package client import ( "context" + "sync" "go.uber.org/zap" + "github.com/rasorp/smuggle/internal/network" "github.com/rasorp/smuggle/internal/types" ) -func (c *Client) startSubnetUpdateHandler() error { - - for _, network := range c.networks { - c.logger.Info("starting network subnet watcher", network.LoggingPairs()...) +type subnetWatcher struct { + cID string + logger *zap.Logger + store types.Store + network *types.Network + networkManager *network.Manager + + // localSubnet is the local subnet allocated for this network on the + // current host. It is used to populate LocalSubnets in SetRemote and + // DeleteRemote calls for policy-based routing. + localSubnet *types.Subnet + + // shutdownCh is the client shutdown channel used to indicate that the agent + // is shutting down and all long-running processes should exit. This is a + // coarse-grained signal. + shutdownCh chan struct{} + + // stopCh is used to signal to this specific subnet watcher that it should + // stop. This allows for more fine-grained control, such as when a subnet is + // removed. + stopCh chan struct{} + + // stopWGFn is a callback function that should be called when the + // subnet watcher has fully stopped. This allows for proper coordination of + // shutdown processes across the client. + stopWGFn func() + + // stopOnce ensures that the stop process is only initiated once, preventing + // potential issues from multiple stop signals. + stopOnce sync.Once +} - req := &types.StoreWatchSubnetsReq{NetworkName: network.Name} +func (s *subnetWatcher) start() { - resp, err := c.store.WatchSubnets(context.Background(), req) - if err != nil { - return err - } + s.logger.Debug("starting remote subnet watcher", zap.String("network_name", s.network.Name)) - go c.subnetUpdateHandlerImpl(resp) + req := &types.StoreWatchSubnetsReq{ + NetworkName: s.network.Name, } - return nil + + // The watch subnets store call currently will only ever return a nil error, + // so we can ignore it here. In the future, if the nvar store implementation + // changes or a new store is added, we may need to handle errors here. + resp, _ := s.store.WatchSubnets(context.Background(), req) + go s.runHandler(resp) } -func (c *Client) subnetUpdateHandlerImpl(req *types.StoreWatchSubnetsResp) { - c.shutdownGroup.Add(1) - defer c.shutdownGroup.Done() +func (s *subnetWatcher) runHandler(req *types.StoreWatchSubnetsResp) { + + // This is a small codebase currently and we known this cannot be nil. In + // the future, if this code is refactored or reused in other contexts, we + // may want to add some additional safety checks or validation. + defer s.stopWGFn() for { select { case err := <-req.ErrorCh: - c.logger.Error("error received from subnet watcher", zap.Error(err)) + s.logger.Error("error received from subnet watcher", zap.Error(err)) case set := <-req.ModifyCh: - c.handleSubnetSet(set) + s.handleSubnetSet(set) case del := <-req.DeleteCh: - c.handleSubnetDelete(del) - case <-c.shutdownCh: - c.logger.Info("shutting down subnet update handler") + s.handleSubnetDelete(del) + case <-s.shutdownCh: + s.logger.Info("shutting down subnet update handler") + return + case <-s.stopCh: + s.logger.Info("stopping subnet update handler") return } } } -func (c *Client) handleSubnetDelete(subnets []*types.Subnet) { +func (s *subnetWatcher) handleSubnetDelete(subnets []*types.Subnet) { for _, subnet := range subnets { - // We may log more than one message, so caputre the pairs here to avoid - // multiple calls to the function and slice allocations. - logPairs := subnet.LoggingPairs() - // If the agent has got an update about itself being expired, the // cluster stability is likely compromised. As the addition is not - // hanled here, we simply skip the deletion attempt as it won't because + // handled here, we simply skip the deletion attempt as it won't because // we don't add local subnets this way. - if subnet.ClientID == c.getID() { - c.logger.Warn("received subnet deletion for local client; skipping", logPairs...) + if subnet.ClientID == s.cID { + s.logger.Warn("received subnet deletion for local client; skipping", + subnet.LoggingPairs()..., + ) continue } - c.logger.Debug("deleting remote network subnet", logPairs...) + s.logger.Debug("deleting remote subnet networking", subnet.LoggingPairs()...) + + var localSubnets []*types.Subnet + if s.localSubnet != nil { + localSubnets = []*types.Subnet{s.localSubnet} + } - _, err := c.networkManager.DeleteRemote(&types.NetworkProviderDeleteRemoteReq{ + _, err := s.networkManager.DeleteRemote(&types.NetworkProviderDeleteRemoteReq{ Subnet: subnet, - LocalSubnets: c.subnets, + LocalSubnets: localSubnets, }) if err != nil { - c.logger.Error("failed to delete remote network subnet", - append(logPairs, zap.Error(err))..., + s.logger.Error("failed to delete remote subnet networking", + append(subnet.LoggingPairs(), zap.Error(err))..., ) } else { - c.logger.Info("successfully deleted remote network subnet", logPairs...) + s.logger.Info("successfully deleted remote subnet networking", subnet.LoggingPairs()...) } } } -func (c *Client) handleSubnetSet(subnets []*types.Subnet) { +func (s *subnetWatcher) handleSubnetSet(subnets []*types.Subnet) { for _, subnet := range subnets { // If the subnet belongs to this host, we do not need to perform the // remote set operation. If we did, it would break the local host subnet // routing. - if subnet.ClientID == c.getID() { + if subnet.ClientID == s.cID { continue } - // We may log more than one message, so caputre the pairs here to avoid - // multiple calls to the function and slice allocations. - logPairs := subnet.LoggingPairs() + s.logger.Debug("setting up remote subnet networking", subnet.LoggingPairs()...) - c.logger.Debug("setting up remote network subnet", logPairs...) + var localSubnets []*types.Subnet + if s.localSubnet != nil { + localSubnets = []*types.Subnet{s.localSubnet} + } - _, err := c.networkManager.SetRemote(&types.NetworkProviderSetRemoteReq{ + _, err := s.networkManager.SetRemote(&types.NetworkProviderSetRemoteReq{ Subnet: subnet, - LocalSubnets: c.subnets, + LocalSubnets: localSubnets, }) if err != nil { - c.logger.Error("failed to set up remote network subnet", - append(logPairs, zap.Error(err))..., + s.logger.Error("failed to set up remote subnet networking", + append(subnet.LoggingPairs(), zap.Error(err))..., ) } else { - c.logger.Info("successfully set up remote network subnet", logPairs...) + s.logger.Info("successfully set up remote subnet networking", subnet.LoggingPairs()...) } } } + +func (s *subnetWatcher) stop() { s.stopOnce.Do(func() { close(s.stopCh) }) } diff --git a/internal/network/firewall/iptables/iptables.go b/internal/network/firewall/iptables/iptables.go index 4354566..3681808 100644 --- a/internal/network/firewall/iptables/iptables.go +++ b/internal/network/firewall/iptables/iptables.go @@ -37,7 +37,7 @@ type Manager struct { logger *zap.Logger } -// New creates a new iptables manager +// NewManager creates a new iptables manager func NewManager(logger *zap.Logger) (types.Firewall, error) { ipt, err := iptables.New() if err != nil { @@ -50,9 +50,128 @@ func NewManager(logger *zap.Logger) (types.Firewall, error) { }, nil } +// CreateForwardRules applies forward rules to iptables +func (m *Manager) CreateForwardRules(network *types.Network) error { + + cidr := network.IPv4.Network.String() + bridgeInterface := network.BridgeInterfaceName() + networkInterface := network.InterfaceName() + + m.logger.Debug("setting up forward rules", + zap.String("network_cidr", cidr), + zap.String("bridge_interface", bridgeInterface), + zap.String("network_interface", networkInterface), + ) + + // Ensure the custom chain exists + if err := m.ensureChain("filter", smuggleForwardChainName); err != nil { + return fmt.Errorf("failed to ensure chain %s: %w", smuggleForwardChainName, err) + } + + // Apply all rules to the Smuggle forward chain but skip the jump rule as + // we'll handle it separately. + for _, rule := range m.forwardRules(cidr, bridgeInterface, networkInterface) { + if rule.chain == forwardChainName { + continue + } + if err := m.applyRule(rule); err != nil { + return fmt.Errorf("failed to apply rule: %w", err) + } + } + + // Ensure jump rule is FIRST in FORWARD chain and before Docker chains. This + // is critical because Docker chains don't have a final ACCEPT, so packets + // that don't match fall through to the DROP policy. + if err := m.ensureJumpRuleFirst("filter", forwardChainName, smuggleForwardChainName); err != nil { + return fmt.Errorf("failed to ensure jump rule is first: %w", err) + } + + m.logger.Info("successfully set up forward rules") + return nil +} + +// DeleteForwardRules removes the forward rules from iptables for the given +// network. +func (m *Manager) DeleteForwardRules(network *types.Network) error { + + networkPairs := network.LoggingPairs() + + m.logger.Debug("deleting forward rules", networkPairs...) + + // Iterate over the rules and delete them. We do not consider errors + // fatal here, as we want to attempt to delete all rules as possible. + for _, rule := range m.forwardRules( + network.IPv4.Network.String(), + network.BridgeInterfaceName(), + network.InterfaceName(), + ) { + if err := m.deleteRule(rule); err != nil { + m.logger.Error("failed to delete forward rule", + append(rule.loggingPairs(), zap.Error(err))..., + ) + } + } + + m.logger.Info("successfully deleted forward rules", networkPairs...) + return nil +} + +// CreateMasqRules applies masquerading rules to iptables +func (m *Manager) CreateMasqRules(network *types.Network, subnet *types.Subnet) error { + + ipv4Network := network.IPv4.Network + ipv4Subnet := subnet.IPv4Network + + m.logger.Debug("setting up masquerading rules", + zap.String("network_cidr", ipv4Network.String()), + zap.String("subnet_cidr", ipv4Subnet.String()), + ) + + // Ensure the custom chain exists + if err := m.ensureChain(natTableName, smugglePostroutingChainName); err != nil { + return fmt.Errorf("failed to ensure chain: %w", err) + } + + // Iterate over the rules and apply them. Any error is considered fatal as + // we need these rules to be in place for proper networking. + for _, rule := range m.masqRules(ipv4Network, ipv4Subnet) { + if err := m.applyRule(rule); err != nil { + return fmt.Errorf("failed to apply rule: %w", err) + } + } + + m.logger.Info("successfully set up masquerading rules", + zap.String("network_cidr", ipv4Network.String()), + zap.String("subnet_cidr", ipv4Subnet.String()), + ) + return nil +} + +// DeleteMasqRules removes masquerading rules from iptables for the given +// network and subnet. +func (m *Manager) DeleteMasqRules(network *types.Network, subnet *types.Subnet) error { + + networkPairs := network.LoggingPairs() + + m.logger.Debug("deleting masquerade rules", networkPairs...) + + // Iterate over the rules and delete them. We do not consider errors fatal + // here, as we want to attempt to delete all rules as possible. + for _, rule := range m.masqRules(network.IPv4.Network, subnet.IPv4Network) { + if err := m.deleteRule(rule); err != nil { + m.logger.Error("failed to delete masquerade rule", + append(rule.loggingPairs(), zap.Error(err))..., + ) + } + } + + m.logger.Info("successfully deleted masquerade rules", networkPairs...) + return nil +} + // masqRules generates the iptables rules for masquerading traffic from the // network subnet to external destinations. -func (i *Manager) masqRules(network *types.IPv4Net, subnet *types.IPv4Net) []rule { +func (m *Manager) masqRules(network *types.IPv4Net, subnet *types.IPv4Net) []rule { rules := []rule{ // Jump from POSTROUTING to our custom chain so we can manage rules // independently in our own chain and perform this before other firewall @@ -69,7 +188,7 @@ func (i *Manager) masqRules(network *types.IPv4Net, subnet *types.IPv4Net) []rul }, } - supportsRandomFully := i.ipt.HasRandomFully() + supportsRandomFully := m.ipt.HasRandomFully() networkString := network.String() subnetString := subnet.String() @@ -108,47 +227,16 @@ func (i *Manager) masqRules(network *types.IPv4Net, subnet *types.IPv4Net) []rul return rules } -// SetupMasqRules applies masquerading rules to iptables -func (i *Manager) SetupMasqRules(network *types.Network, subnet *types.Subnet) error { - - ipv4Network := network.IPv4.Network - ipv4Subnet := subnet.IPv4Network - - i.logger.Debug("setting up masquerading rules", - zap.String("network_cidr", ipv4Network.String()), - zap.String("subnet_cidr", ipv4Subnet.String()), - ) - - // Ensure the custom chain exists - if err := i.ensureChain(natTableName, smugglePostroutingChainName); err != nil { - return fmt.Errorf("failed to ensure chain: %w", err) - } - - // Iterate over the rules and apply them. Any error is considered fatal as - // we need these rules to be in place for proper networking. - for _, rule := range i.masqRules(ipv4Network, ipv4Subnet) { - if err := i.applyRule(rule); err != nil { - return fmt.Errorf("failed to apply rule: %w", err) - } - } - - i.logger.Info("successfully set up masquerading rules", - zap.String("network_cidr", ipv4Network.String()), - zap.String("subnet_cidr", ipv4Subnet.String()), - ) - return nil -} - // ensureChain ensures an iptables chain exists, creating it if necessary -func (i *Manager) ensureChain(table, chain string) error { - chains, err := i.ipt.ListChains(table) +func (m *Manager) ensureChain(table, chain string) error { + chains, err := m.ipt.ListChains(table) if err != nil { return fmt.Errorf("failed to list chains: %w", err) } // Check if chain already exists if slices.Contains(chains, chain) { - i.logger.Debug("chain already exists, skipping creation", + m.logger.Debug("chain already exists, skipping creation", zap.String("table", table), zap.String("chain", chain), ) @@ -156,16 +244,16 @@ func (i *Manager) ensureChain(table, chain string) error { } // Create the chain - i.logger.Debug("creating iptables chain", + m.logger.Debug("creating iptables chain", zap.String("table", table), zap.String("chain", chain), ) - if err := i.ipt.NewChain(table, chain); err != nil { + if err := m.ipt.NewChain(table, chain); err != nil { return fmt.Errorf("failed to create chain: %w", err) } - i.logger.Info("successfully created chain", + m.logger.Info("successfully created chain", zap.String("table", table), zap.String("chain", chain), ) @@ -174,8 +262,9 @@ func (i *Manager) ensureChain(table, chain string) error { } // applyRule ensures an iptables rule exists, adding it if necessary -func (i *Manager) applyRule(rule rule) error { - exists, err := i.ipt.Exists(rule.table, rule.chain, rule.spec...) +func (m *Manager) applyRule(rule rule) error { + + exists, err := m.ipt.Exists(rule.table, rule.chain, rule.spec...) if err != nil { return fmt.Errorf("failed to check if rule exists: %w", err) } @@ -185,15 +274,43 @@ func (i *Manager) applyRule(rule rule) error { loggingPairs := rule.loggingPairs() if !exists { - i.logger.Debug("applying iptables rule", loggingPairs...) + m.logger.Debug("applying iptables rule", loggingPairs...) - if err := i.ipt.Append(rule.table, rule.chain, rule.spec...); err != nil { + if err := m.ipt.Append(rule.table, rule.chain, rule.spec...); err != nil { return fmt.Errorf("failed to apply rule: %w", err) } - i.logger.Info("successfully applied iptables rule", loggingPairs...) + m.logger.Info("successfully applied iptables rule", loggingPairs...) + } else { + m.logger.Debug("iptables rule already exists, skipping apply", loggingPairs...) + } + + return nil +} + +// deleteRule removes an iptables rule if it exists. If it does not exist, it +// is a no-op. +func (m *Manager) deleteRule(rule rule) error { + + exists, err := m.ipt.Exists(rule.table, rule.chain, rule.spec...) + if err != nil { + return fmt.Errorf("failed to check if rule exists: %w", err) + } + + // Generate the logging pairs once, that will be used in both branches and + // potentially multiple times. + loggingPairs := rule.loggingPairs() + + if exists { + m.logger.Debug("deleting iptables rule", loggingPairs...) + + if err := m.ipt.Delete(rule.table, rule.chain, rule.spec...); err != nil { + return fmt.Errorf("failed to delete rule: %w", err) + } + + m.logger.Info("successfully deleted iptables rule", loggingPairs...) } else { - i.logger.Debug("iptables rule already exists, skipping apply", loggingPairs...) + m.logger.Debug("iptables rule does not exist, skipping delete", loggingPairs...) } return nil @@ -201,7 +318,7 @@ func (i *Manager) applyRule(rule rule) error { // forwardRules generates iptables rules for forwarding traffic that allows // traffic to be forwarded to and from the network range. -func (i *Manager) forwardRules(networkCIDR, bridgeInterface, networkInterface string) []rule { +func (m *Manager) forwardRules(networkCIDR, bridgeInterface, networkInterface string) []rule { return []rule{ // Jump to custom chain to manage forward rules independently. This // ensures Smuggle rules are evaluated before other node firewall rules. @@ -308,71 +425,31 @@ func (i *Manager) forwardRules(networkCIDR, bridgeInterface, networkInterface st } } -// SetupForwardRules applies forward rules to iptables -func (i *Manager) SetupForwardRules(network *types.Network) error { - - cidr := network.IPv4.Network.String() - bridgeInterface := network.BridgeInterfaceName() - networkInterface := network.InterfaceName() - - i.logger.Debug("setting up forward rules", - zap.String("network_cidr", cidr), - zap.String("bridge_interface", bridgeInterface), - zap.String("network_interface", networkInterface), - ) - - // Ensure the custom chain exists - if err := i.ensureChain("filter", smuggleForwardChainName); err != nil { - return fmt.Errorf("failed to ensure chain %s: %w", smuggleForwardChainName, err) - } - - // Apply all rules to the Smuggle forward chain but Skip the jump rule as - // we'll handle it separately. - for _, rule := range i.forwardRules(cidr, bridgeInterface, networkInterface) { - if rule.chain == forwardChainName { - continue - } - if err := i.applyRule(rule); err != nil { - return fmt.Errorf("failed to apply rule: %w", err) - } - } - - // Ensure jump rule is FIRST in FORWARD chain and before Docker chains. This - // is critical because Docker chains don't have a final ACCEPT, so packets - // that don't match fall through to the DROP policy. - if err := i.ensureJumpRuleFirst("filter", forwardChainName, smuggleForwardChainName); err != nil { - return fmt.Errorf("failed to ensure jump rule is first: %w", err) - } - - i.logger.Info("successfully set up forward rules") - return nil -} - -// ensureJumpRuleFirst ensures a jump rule exists and is at position 1 -// This is necessary to ensure Smuggle rules run before Docker's chains -func (i *Manager) ensureJumpRuleFirst(table, chain, targetChain string) error { +// ensureJumpRuleFirst ensures a jump rule exists and is at position 1. +// This is necessary to ensure Smuggle rules run before Docker's chains. +func (m *Manager) ensureJumpRuleFirst(table, chain, targetChain string) error { ruleSpec := []string{"-m", "comment", "--comment", "smuggle forward", "-j", targetChain} // Check if rule exists - exists, err := i.ipt.Exists(table, chain, ruleSpec...) + exists, err := m.ipt.Exists(table, chain, ruleSpec...) if err != nil { return fmt.Errorf("failed to check if jump rule exists: %w", err) } if exists { // Rule exists but might not be first. Delete and re-insert. - if err := i.ipt.Delete(table, chain, ruleSpec...); err != nil { - i.logger.Warn("failed to delete existing jump rule, will try to insert anyway", + if err := m.ipt.Delete(table, chain, ruleSpec...); err != nil { + m.logger.Warn("failed to delete existing jump rule, will try to insert anyway", zap.Error(err)) } } // Insert at position 1 (first rule, before Docker chains) - if err := i.ipt.Insert(table, chain, 1, ruleSpec...); err != nil { + if err := m.ipt.Insert(table, chain, 1, ruleSpec...); err != nil { return fmt.Errorf("failed to insert jump rule at position 1: %w", err) } - i.logger.Info("ensured jump rule is first in chain", + m.logger.Info("ensured jump rule is first in chain", zap.String("table", table), zap.String("chain", chain), zap.String("target", targetChain), @@ -381,25 +458,22 @@ func (i *Manager) ensureJumpRuleFirst(table, chain, targetChain string) error { return nil } -// EnsureIsolation creates REJECT rules to prevent cross-network communication. +// CreateIsolation creates REJECT rules to prevent cross-network communication. // For each pair of networks, it creates rules that reject traffic from one // network's interfaces to another network's interfaces. This uses the + // wildcard to match all interfaces belonging to a network (both bridge and // VXLAN). -func (i *Manager) EnsureIsolation(networks []*types.Network) error { +func (m *Manager) CreateIsolation(networks []*types.Network) error { // There is no need to apply isolation rules if there are less than 2 // networks. if len(networks) < 2 { - i.logger.Debug("no isolation rules needed", zap.Int("network_count", len(networks))) + m.logger.Debug("no isolation rules needed") return nil } - i.logger.Info("ensuring network isolation", - zap.Int("network_count", len(networks))) - // Ensure the custom chain exists - if err := i.ensureChain("filter", smuggleForwardChainName); err != nil { + if err := m.ensureChain("filter", smuggleForwardChainName); err != nil { return fmt.Errorf("failed to ensure chain: %w", err) } @@ -408,7 +482,6 @@ func (i *Manager) EnsureIsolation(networks []*types.Network) error { // For each network, create REJECT rules to all other networks for _, sourceNetwork := range networks { - sourcePrefix := sourceNetwork.Name + "+" for _, destNetwork := range networks { // Skip if same network @@ -416,69 +489,109 @@ func (i *Manager) EnsureIsolation(networks []*types.Network) error { continue } - destPrefix := destNetwork.Name + "+" - // Create REJECT rule for this network pair. The + wildcard matches // both bridge and VXLAN interfaces. - isolationRules = append(isolationRules, rule{ - id: fmt.Sprintf("reject-%s-to-%s", sourceNetwork.Name, destNetwork.Name), - table: "filter", - chain: smuggleForwardChainName, - spec: []string{ - "-i", sourcePrefix, - "-o", destPrefix, - "-m", "comment", - "--comment", fmt.Sprintf("smuggle isolate %s from %s", sourceNetwork.Name, destNetwork.Name), - "-j", "REJECT", - "--reject-with", "icmp-net-prohibited", - }, - }) + isolationRules = append( + isolationRules, + m.isolationRule(sourceNetwork.Name+"+", destNetwork.Name+"+"), + ) } } - i.logger.Debug("applying isolation rules", - zap.Int("rule_count", len(isolationRules))) + m.logger.Debug("creating isolation rules") - // Apply all isolation rules - // These need to be inserted near the beginning of the chain, right after - // ESTABLISHED,RELATED but before any ACCEPT rules for _, rule := range isolationRules { - if err := i.ensureIsolationRule(rule); err != nil { - return fmt.Errorf("failed to apply isolation rule: %w", err) + if err := m.ensureIsolationRule(rule); err != nil { + return fmt.Errorf("failed to create isolation rule: %w", err) } } - i.logger.Info("successfully ensured network isolation", - zap.Int("network_count", len(networks)), - zap.Int("rule_count", len(isolationRules))) - + m.logger.Info("successfully created isolation rules") return nil } // ensureIsolationRule ensures an isolation REJECT rule exists in the chain. // Unlike applyRule which appends, this inserts the rule at a specific position // to ensure isolation rules run before ACCEPT rules. -func (i *Manager) ensureIsolationRule(rule rule) error { - exists, err := i.ipt.Exists(rule.table, rule.chain, rule.spec...) +func (m *Manager) ensureIsolationRule(rule rule) error { + exists, err := m.ipt.Exists(rule.table, rule.chain, rule.spec...) if err != nil { return fmt.Errorf("failed to check if rule exists: %w", err) } - loggingPairs := rule.loggingPairs() - if !exists { - i.logger.Debug("inserting isolation rule", loggingPairs...) + m.logger.Debug("creating isolation rule", rule.loggingPairs()...) - // Insert at position 2 (right after ESTABLISHED,RELATED which is at position 1) - // This ensures isolation rules run before any ACCEPT rules - if err := i.ipt.Insert(rule.table, rule.chain, 2, rule.spec...); err != nil { - return fmt.Errorf("failed to insert isolation rule: %w", err) + // Insert at position 2 which is immediately after ESTABLISHED,RELATED + // which is at position 1. This ensures isolation rules run before any + // ACCEPT rules. + if err := m.ipt.Insert(rule.table, rule.chain, 2, rule.spec...); err != nil { + return fmt.Errorf("failed to create isolation rule: %w", err) } + } - i.logger.Info("successfully inserted isolation rule", loggingPairs...) - } else { - i.logger.Debug("isolation rule already exists, skipping", loggingPairs...) + return nil +} + +// DeleteIsolation removes the isolation rules for the deleted networks, while +// keeping the rules for the existing networks intact. +func (m *Manager) DeleteIsolation(exist, deleted []*types.Network) error { + + if len(deleted) == 0 { + m.logger.Debug("no networks to delete isolation rules for") + return nil } + m.logger.Debug("deleting network isolation rules") + + // Build a combined list of all networks (existing + deleted) to check against + allNetworks := append([]*types.Network{}, exist...) + allNetworks = append(allNetworks, deleted...) + + // For each deleted network, remove all isolation rules where it appears + // as either source or destination + for _, deletedNetwork := range deleted { + deletedPrefix := deletedNetwork.Name + "+" + + for _, otherNetwork := range allNetworks { + + // Skip if same network. + if deletedNetwork.Name == otherNetwork.Name { + continue + } + + otherPrefix := otherNetwork.Name + "+" + + // Delete rule where deleted network is the SOURCE (deleted -> other) + if err := m.deleteRule(m.isolationRule(deletedPrefix, otherPrefix)); err != nil { + return fmt.Errorf("failed to delete isolation rule: %w", err) + } + + // Delete rule where deleted network is the DESTINATION (other -> deleted) + if err := m.deleteRule(m.isolationRule(otherPrefix, deletedPrefix)); err != nil { + return fmt.Errorf("failed to delete isolation rule: %w", err) + } + } + } + + m.logger.Info("successfully deleted network isolation rules") return nil } + +// isolationRule generates the isolation REJECT rule for the given source and +// destination interface prefixes. +func (m *Manager) isolationRule(src, dst string) rule { + return rule{ + id: fmt.Sprintf("reject-%s-to-%s", src, dst), + table: "filter", + chain: smuggleForwardChainName, + spec: []string{ + "-i", src, + "-o", dst, + "-m", "comment", + "--comment", fmt.Sprintf("smuggle isolate %s from %s", src, dst), + "-j", "REJECT", + "--reject-with", "icmp-net-prohibited", + }, + } +} diff --git a/internal/network/network.go b/internal/network/network.go index fe59804..c3b844e 100644 --- a/internal/network/network.go +++ b/internal/network/network.go @@ -52,6 +52,18 @@ func NewManager(logger *zap.Logger, intf string) (*Manager, error) { return &m, nil } +func (m *Manager) DeleteLocal( + req *types.NetworkProviderDeleteLocalReq, +) (*types.NetworkProviderDeleteLocalResp, error) { + + provider, ok := m.providers[req.Subnet.Provider] + if !ok { + return nil, fmt.Errorf("unknown network provider %q", req.Subnet.Provider) + } + + return provider.DeleteLocal(req) +} + func (m *Manager) SetLocal( req *types.NetworkProviderSetReq, ) (*types.NetworkProviderSetResp, error) { diff --git a/internal/network/provider/vxlan/vxlan_ipv4_linux.go b/internal/network/provider/vxlan/vxlan_ipv4_linux.go index f33f823..d9a1fa4 100644 --- a/internal/network/provider/vxlan/vxlan_ipv4_linux.go +++ b/internal/network/provider/vxlan/vxlan_ipv4_linux.go @@ -70,3 +70,21 @@ func (p *Provider) createIPv4( return vxlanLink, nil } + +// deleteIPv4 deletes the IPv4 interface associated with the provided subnet. +func (p *Provider) deleteIPv4(cfg *types.Subnet) error { + + link, err := netlink.LinkByName(cfg.InterfaceName()) + if err != nil { + if _, ok := err.(netlink.LinkNotFoundError); ok { + return nil + } + return fmt.Errorf("failed to get vxlan link: %w", err) + } + + if err := netlink.LinkDel(link); err != nil { + return fmt.Errorf("failed to delete vxlan link: %w", err) + } + + return nil +} diff --git a/internal/network/provider/vxlan/vxlan_linux.go b/internal/network/provider/vxlan/vxlan_linux.go index c1b2a17..0ab704d 100644 --- a/internal/network/provider/vxlan/vxlan_linux.go +++ b/internal/network/provider/vxlan/vxlan_linux.go @@ -53,6 +53,19 @@ func New(logger *zap.Logger) types.NetworkProvider { func (p *Provider) Name() string { return providerName } +func (p *Provider) DeleteLocal( + req *types.NetworkProviderDeleteLocalReq, +) (*types.NetworkProviderDeleteLocalResp, error) { + + if err := p.deleteIPv4(req.Subnet); err != nil { + return nil, err + } + + p.logger.Info("successfully deleted local VXLAN interface", req.Subnet.LoggingPairs()...) + + return &types.NetworkProviderDeleteLocalResp{}, nil +} + func (p *Provider) SetLocal( req *types.NetworkProviderSetReq, ) (*types.NetworkProviderSetResp, error) { @@ -91,7 +104,7 @@ func (p *Provider) SetLocal( return nil, fmt.Errorf("failed to marshal vxlan config: %v", err) } - p.logger.Info("setup local VXLAN interface", cfg.loggingPairs()...) + p.logger.Info("successfully set up local VXLAN interface", cfg.loggingPairs()...) // Create a copy of the subnet to avoid mutating the request object and // ensure we don't accidentally modify the caller's data. diff --git a/internal/store/file/file.go b/internal/store/file/file.go index c63908c..37fc238 100644 --- a/internal/store/file/file.go +++ b/internal/store/file/file.go @@ -96,3 +96,12 @@ func (s *CNIStore) Set(cfg *types.CNIConfig) error { return nil } + +func (s *CNIStore) Delete(name string) error { + if err := os.Remove(filepath.Join(s.path, name+".conf")); err != nil { + if !errors.Is(err, os.ErrNotExist) { + return fmt.Errorf("failed to delete CNI config file: %w", err) + } + } + return nil +} diff --git a/internal/types/cni.go b/internal/types/cni.go index 1363145..d8f8822 100644 --- a/internal/types/cni.go +++ b/internal/types/cni.go @@ -1,6 +1,7 @@ package types type CNIStore interface { + Delete(name string) error Set(*CNIConfig) error } diff --git a/internal/types/firewall.go b/internal/types/firewall.go index b7c004a..a5a95f5 100644 --- a/internal/types/firewall.go +++ b/internal/types/firewall.go @@ -5,19 +5,35 @@ package types // allow traffic forwarding and masquerading for networks and subnets. type Firewall interface { - // EnsureIsolation ensures that all networks in the provided list are + // CreateIsolation ensures that all networks in the provided list are // isolated from each other by adding REJECT rules for cross-network // traffic. This prevents containers on different networks from // communicating with each other. - EnsureIsolation([]*Network) error + CreateIsolation([]*Network) error - // SetupForwardRules sets up firewall forwarding rules for the provided + // DeleteIsolation removes the isolation rules for the deleted networks, + // while keeping the rules for the existing networks intact. It allows the + // agent to clean up any rules that are no longer needed after a network is + // deleted. + DeleteIsolation(exist, deleted []*Network) error + + // CreateForwardRules sets up firewall forwarding rules for the provided // network. This is used to allow traffic to be forwarded between subnets on // the network. - SetupForwardRules(*Network) error + CreateForwardRules(network *Network) error + + // DeleteForwardRules deletes firewall forwarding rules for the provided + // network. This is used to clean up rules when a network is deleted from + // the store. + DeleteForwardRules(network *Network) error - // SetupMasqRules sets up firewall masquerading rules for the provided + // CreateMasqRules sets up firewall masquerading rules for the provided // network and subnet. This is used to enable NAT for traffic leaving the // subnet to external destinations. - SetupMasqRules(*Network, *Subnet) error + CreateMasqRules(network *Network, subnet *Subnet) error + + // DeleteMasqRules deletes firewall masquerading rules for the provided + // network and subnet. This is used to clean up rules when a subnet is + // deleted from the store. + DeleteMasqRules(network *Network, subnet *Subnet) error } diff --git a/internal/types/network.go b/internal/types/network.go index 0286a03..cb19426 100644 --- a/internal/types/network.go +++ b/internal/types/network.go @@ -1,6 +1,7 @@ package types import ( + "bytes" "encoding/json" "errors" "fmt" @@ -58,6 +59,66 @@ func (n *Network) Canonicalize() { } } +// Equals compares two network configurations for equality. The name field is +// intentionally excluded from the comparison as networks are identified by +// this field. +func (n *Network) Equals(other *Network) bool { + + if n == nil && other == nil { + return true + } + if n == nil || other == nil { + return false + } + + if (n.IPMasq == nil) != (other.IPMasq == nil) { + return false + } + if n.IPMasq != nil && other.IPMasq != nil && *n.IPMasq != *other.IPMasq { + return false + } + + // Compare IPv4 configuration. + if (n.IPv4 == nil) != (other.IPv4 == nil) { + return false + } + if n.IPv4 != nil && other.IPv4 != nil { + if (n.IPv4.Network == nil) != (other.IPv4.Network == nil) { + return false + } + if n.IPv4.Network != nil && other.IPv4.Network != nil { + if n.IPv4.Network.IP != other.IPv4.Network.IP || + n.IPv4.Network.Size != other.IPv4.Network.Size { + return false + } + } + if n.IPv4.Min != other.IPv4.Min { + return false + } + if n.IPv4.Max != other.IPv4.Max { + return false + } + if n.IPv4.Size != other.IPv4.Size { + return false + } + } + + // Compare Provider configuration. + if (n.Provider == nil) != (other.Provider == nil) { + return false + } + if n.Provider != nil && other.Provider != nil { + if n.Provider.Name != other.Provider.Name { + return false + } + if !bytes.Equal(n.Provider.Config, other.Provider.Config) { + return false + } + } + + return true +} + // InterfaceName returns the name of the network interface that is used for // this network. Networks are expected to have a single interface per host, so // this with a static suffix of "0" is sufficient to be unique. diff --git a/internal/types/provider.go b/internal/types/provider.go index 5d2c4f6..c030db0 100644 --- a/internal/types/provider.go +++ b/internal/types/provider.go @@ -11,6 +11,11 @@ type NetworkProvider interface { // Name returns the unique identifier for this provider. Name() string + // DeleteLocal removes the local subnet configuration from this host. This + // is called when a network has been removed and we need to clean up local + // resources. + DeleteLocal(req *NetworkProviderDeleteLocalReq) (*NetworkProviderDeleteLocalResp, error) + // SetLocal configures the local subnet for this host. SetLocal(*NetworkProviderSetReq) (*NetworkProviderSetResp, error) @@ -21,6 +26,12 @@ type NetworkProvider interface { SetRemote(*NetworkProviderSetRemoteReq) (*NetworkProviderSetRemoteResp, error) } +type NetworkProviderDeleteLocalReq struct { + Subnet *Subnet +} + +type NetworkProviderDeleteLocalResp struct{} + // NetworkProviderSetReq contains parameters for setting up a local subnet. type NetworkProviderSetReq struct { HostInterface *net.Interface