X-Git-Url: http://git.sameswireless.fr/l2tpns.git/blobdiff_plain/4f253feef08f01c03bc63f70f9de3937541ccd79..refs/heads/infinite-loop-protection:/cluster.c?ds=inline diff --git a/cluster.c b/cluster.c index 41dc10f..d9b226f 100644 --- a/cluster.c +++ b/cluster.c @@ -1,7 +1,5 @@ // L2TPNS Clustering Stuff -char const *cvs_id_cluster = "$Id: cluster.c,v 1.44 2005/06/28 14:48:19 bodea Exp $"; - #include #include #include @@ -19,10 +17,12 @@ char const *cvs_id_cluster = "$Id: cluster.c,v 1.44 2005/06/28 14:48:19 bodea Ex #include #include +#include "dhcp6.h" #include "l2tpns.h" #include "cluster.h" #include "util.h" #include "tbf.h" +#include "pppoe.h" #ifdef BGP #include "bgp.h" @@ -42,6 +42,7 @@ extern int cluster_sockfd; // The filedescriptor for the cluster communications in_addr_t my_address = 0; // The network address of my ethernet port. static int walk_session_number = 0; // The next session to send when doing the slow table walk. +static int walk_bundle_number = 0; // The next bundle to send when doing the slow table walk. static int walk_tunnel_number = 0; // The next tunnel to send when doing the slow table walk. int forked = 0; // Sanity check: CLI must not diddle with heartbeat table @@ -56,7 +57,7 @@ static struct { static struct { int seq; int size; - char data[MAX_HEART_SIZE]; + uint8_t data[MAX_HEART_SIZE]; } past_hearts[HB_HISTORY_SIZE]; // Ring buffer of heartbeats that we've recently sent out. Needed so // we can re-transmit if needed. @@ -85,6 +86,7 @@ int cluster_init() int opt; config->cluster_undefined_sessions = MAXSESSION-1; + config->cluster_undefined_bundles = MAXBUNDLE-1; config->cluster_undefined_tunnels = MAXTUNNEL-1; if (!config->cluster_address) @@ -96,7 +98,7 @@ int cluster_init() memset(&addr, 0, sizeof(addr)); addr.sin_family = AF_INET; - addr.sin_port = htons(CLUSTERPORT); + addr.sin_port = htons(config->cluster_port); addr.sin_addr.s_addr = INADDR_ANY; setsockopt(cluster_sockfd, SOL_SOCKET, SO_REUSEADDR, &addr, sizeof(addr)); @@ -127,6 +129,15 @@ int cluster_init() opt = 0; // Turn off multicast loopback. setsockopt(cluster_sockfd, IPPROTO_IP, IP_MULTICAST_LOOP, &opt, sizeof(opt)); + if (config->cluster_mcast_ttl != 1) + { + uint8_t ttl = 0; + if (config->cluster_mcast_ttl > 0) + ttl = config->cluster_mcast_ttl < 256 ? config->cluster_mcast_ttl : 255; + + setsockopt(cluster_sockfd, IPPROTO_IP, IP_MULTICAST_TTL, &ttl, sizeof(ttl)); + } + if (setsockopt(cluster_sockfd, IPPROTO_IP, IP_ADD_MEMBERSHIP, &mreq, sizeof(mreq)) < 0) { LOG(0, 0, 0, "Failed to setsockopt (join mcast group): %s\n", strerror(errno)); @@ -159,7 +170,7 @@ static int cluster_send_data(void *data, int datalen) if (!config->cluster_address) return 0; addr.sin_addr.s_addr = config->cluster_address; - addr.sin_port = htons(CLUSTERPORT); + addr.sin_port = htons(config->cluster_port); addr.sin_family = AF_INET; LOG(5, 0, 0, "Cluster send data: %d bytes\n", datalen); @@ -178,7 +189,7 @@ static int cluster_send_data(void *data, int datalen) // Maintains the format. Assumes that the caller // has passed in a big enough buffer! // -static void add_type(char **p, int type, int more, char *data, int size) +static void add_type(uint8_t **p, int type, int more, uint8_t *data, int size) { *((uint32_t *) (*p)) = type; *p += sizeof(uint32_t); @@ -218,7 +229,7 @@ static void cluster_uptodate(void) if (config->cluster_iam_uptodate) return; - if (config->cluster_undefined_sessions || config->cluster_undefined_tunnels) + if (config->cluster_undefined_sessions || config->cluster_undefined_tunnels || config->cluster_undefined_bundles) return; config->cluster_iam_uptodate = 1; @@ -231,7 +242,7 @@ static void cluster_uptodate(void) // Send a unicast UDP packet to a peer with 'data' as the // contents. // -static int peer_send_data(in_addr_t peer, char *data, int size) +static int peer_send_data(in_addr_t peer, uint8_t *data, int size) { struct sockaddr_in addr = {0}; @@ -242,7 +253,7 @@ static int peer_send_data(in_addr_t peer, char *data, int size) return -1; addr.sin_addr.s_addr = peer; - addr.sin_port = htons(CLUSTERPORT); + addr.sin_port = htons(config->cluster_port); addr.sin_family = AF_INET; LOG_HEX(5, "Peer send", data, size); @@ -259,10 +270,10 @@ static int peer_send_data(in_addr_t peer, char *data, int size) // // Send a structured message to a peer with a single element of type 'type'. // -static int peer_send_message(in_addr_t peer, int type, int more, char *data, int size) +static int peer_send_message(in_addr_t peer, int type, int more, uint8_t *data, int size) { - char buf[65536]; // Vast overkill. - char *p = buf; + uint8_t buf[65536]; // Vast overkill. + uint8_t *p = buf; LOG(4, 0, 0, "Sending message to peer (type %d, more %d, size %d)\n", type, more, size); add_type(&p, type, more, data, size); @@ -271,10 +282,10 @@ static int peer_send_message(in_addr_t peer, int type, int more, char *data, int } // send a packet to the master -static int _forward_packet(char *data, int size, in_addr_t addr, int port, int type) +static int _forward_packet(uint8_t *data, int size, in_addr_t addr, int port, int type) { - char buf[65536]; // Vast overkill. - char *p = buf; + uint8_t buf[65536]; // Vast overkill. + uint8_t *p = buf; if (!config->cluster_master_address) // No election has been held yet. Just skip it. return -1; @@ -282,7 +293,7 @@ static int _forward_packet(char *data, int size, in_addr_t addr, int port, int t LOG(4, 0, 0, "Forwarding packet from %s to master (size %d)\n", fmtaddr(addr, 0), size); STAT(c_forwarded); - add_type(&p, type, addr, (char *) &port, sizeof(port)); // ick. should be uint16_t + add_type(&p, type, addr, (uint8_t *) &port, sizeof(port)); // ick. should be uint16_t memcpy(p, data, size); p += size; @@ -294,14 +305,44 @@ static int _forward_packet(char *data, int size, in_addr_t addr, int port, int t // // The master just processes the payload as if it had // received it off the tun device. -// -int master_forward_packet(char *data, int size, in_addr_t addr, int port) +//(note: THIS ROUTINE WRITES TO pack[-6]). +int master_forward_packet(uint8_t *data, int size, in_addr_t addr, uint16_t port, uint16_t indexudp) +{ + uint8_t *p = data - (3 * sizeof(uint32_t)); + uint8_t *psave = p; + uint32_t indexandport = port | ((indexudp << 16) & 0xFFFF0000); + + if (!config->cluster_master_address) // No election has been held yet. Just skip it. + return -1; + + LOG(4, 0, 0, "Forwarding packet from %s to master (size %d)\n", fmtaddr(addr, 0), size); + + STAT(c_forwarded); + add_type(&p, C_FORWARD, addr, (uint8_t *) &indexandport, sizeof(indexandport)); + + return peer_send_data(config->cluster_master_address, psave, size + (3 * sizeof(uint32_t))); +} + +// Forward PPPOE packet to the master. +//(note: THIS ROUTINE WRITES TO pack[-4]). +int master_forward_pppoe_packet(uint8_t *data, int size, uint8_t codepad) { - return _forward_packet(data, size, addr, port, C_FORWARD); + uint8_t *p = data - (2 * sizeof(uint32_t)); + uint8_t *psave = p; + + if (!config->cluster_master_address) // No election has been held yet. Just skip it. + return -1; + + LOG(4, 0, 0, "Forward PPPOE packet to master, code %s (size %d)\n", get_string_codepad(codepad), size); + + STAT(c_forwarded); + add_type(&p, C_PPPOE_FORWARD, codepad, NULL, 0); + + return peer_send_data(config->cluster_master_address, psave, size + (2 * sizeof(uint32_t))); } // Forward a DAE RADIUS packet to the master. -int master_forward_dae_packet(char *data, int size, in_addr_t addr, int port) +int master_forward_dae_packet(uint8_t *data, int size, in_addr_t addr, int port) { return _forward_packet(data, size, addr, port, C_FORWARD_DAE); } @@ -313,10 +354,10 @@ int master_forward_dae_packet(char *data, int size, in_addr_t addr, int port) // token bucket queue, and lets normal processing take care // of it. // -int master_throttle_packet(int tbfid, char *data, int size) +int master_throttle_packet(int tbfid, uint8_t *data, int size) { - char buf[65536]; // Vast overkill. - char *p = buf; + uint8_t buf[65536]; // Vast overkill. + uint8_t *p = buf; if (!config->cluster_master_address) // No election has been held yet. Just skip it. return -1; @@ -338,10 +379,10 @@ int master_throttle_packet(int tbfid, char *data, int size) // // (Note that this must be called with the tun header // as the start of the data). -int master_garden_packet(sessionidt s, char *data, int size) +int master_garden_packet(sessionidt s, uint8_t *data, int size) { - char buf[65536]; // Vast overkill. - char *p = buf; + uint8_t buf[65536]; // Vast overkill. + uint8_t *p = buf; if (!config->cluster_master_address) // No election has been held yet. Just skip it. return -1; @@ -354,11 +395,33 @@ int master_garden_packet(sessionidt s, char *data, int size) } +// +// Forward a MPPP packet to the master for handling. +// +// (Note that this must be called with the tun header +// as the start of the data). +// (i.e. this routine writes to data[-8]). +int master_forward_mppp_packet(sessionidt s, uint8_t *data, int size) +{ + uint8_t *p = data - (2 * sizeof(uint32_t)); + uint8_t *psave = p; + + if (!config->cluster_master_address) // No election has been held yet. Just skip it. + return -1; + + LOG(4, 0, 0, "Forward MPPP packet to master (size %d)\n", size); + + add_type(&p, C_MPPP_FORWARD, s, NULL, 0); + + return peer_send_data(config->cluster_master_address, psave, size + (2 * sizeof(uint32_t))); + +} + // // Send a chunk of data as a heartbeat.. // We save it in the history buffer as we do so. // -static void send_heartbeat(int seq, char *data, int size) +static void send_heartbeat(int seq, uint8_t *data, int size) { int i; @@ -380,8 +443,8 @@ static void send_heartbeat(int seq, char *data, int size) // void cluster_send_ping(time_t basetime) { - char buff[100 + sizeof(pingt)]; - char *p = buff; + uint8_t buff[100 + sizeof(pingt)]; + uint8_t *p = buff; pingt x; if (config->cluster_iam_master && basetime) // We're heartbeating so no need to ping. @@ -391,10 +454,10 @@ void cluster_send_ping(time_t basetime) x.ver = 1; x.addr = config->bind_address; - x.undef = config->cluster_undefined_sessions + config->cluster_undefined_tunnels; + x.undef = config->cluster_undefined_sessions + config->cluster_undefined_tunnels + config->cluster_undefined_bundles; x.basetime = basetime; - add_type(&p, C_PING, basetime, (char *) &x, sizeof(x)); + add_type(&p, C_PING, basetime, (uint8_t *) &x, sizeof(x)); cluster_send_data(buff, (p-buff) ); } @@ -417,12 +480,6 @@ void master_update_counts(void) if (!config->cluster_master_address) // If we don't have a master, skip it for a while. return; - // C_BYTES format changed in 2.1.0 (cluster version 5) - // during upgrade from previous versions, hang onto our counters - // for a bit until the new master comes up - if (config->cluster_last_hb_ver < 5) - return; - i = MAX_B_RECS * 5; // Examine max 3000 sessions; if (config->cluster_highest_sessionid > i) i = config->cluster_highest_sessionid; @@ -456,7 +513,7 @@ void master_update_counts(void) // Forward the data to the master. LOG(4, 0, 0, "Sending byte counters to master (%d elements)\n", c); - peer_send_message(config->cluster_master_address, C_BYTES, c, (char *) &b, sizeof(b[0]) * c); + peer_send_message(config->cluster_master_address, C_BYTES, c, (uint8_t *) &b, sizeof(b[0]) * c); return; } @@ -511,7 +568,7 @@ void cluster_check_slaves(void) // void cluster_check_master(void) { - int i, count, tcount, high_unique_id = 0; + int i, count, high_unique_id = 0; int last_free = 0; clockt t = TIME; static int probed = 0; @@ -581,6 +638,7 @@ void cluster_check_master(void) // to become a master!!! config->cluster_iam_master = 1; + pppoe_send_garp(); // gratuitous arp of the pppoe interface LOG(0, 0, 0, "I am declaring myself the master!\n"); @@ -597,7 +655,7 @@ void cluster_check_master(void) // Count the highest used tunnel number as well. // config->cluster_highest_tunnelid = 0; - for (i = 0, tcount = 0; i < MAXTUNNEL; ++i) { + for (i = 0; i < MAXTUNNEL; ++i) { if (tunnel[i].state == TUNNELUNDEF) tunnel[i].state = TUNNELFREE; @@ -605,6 +663,19 @@ void cluster_check_master(void) config->cluster_highest_tunnelid = i; } + // + // Go through and mark all the bundles as defined. + // Count the highest used bundle number as well. + // + config->cluster_highest_bundleid = 0; + for (i = 0; i < MAXBUNDLE; ++i) { + if (bundle[i].state == BUNDLEUNDEF) + bundle[i].state = BUNDLEFREE; + + if (bundle[i].state != BUNDLEFREE && i > config->cluster_highest_bundleid) + config->cluster_highest_bundleid = i; + } + // // Go through and mark all the sessions as being defined. // reset the idle timeouts. @@ -630,7 +701,7 @@ void cluster_check_master(void) } // Reset idle timeouts.. - session[i].last_packet = time_now; + session[i].last_packet = session[i].last_data = time_now; // Reset die relative to our uptime rather than the old master's if (session[i].die) session[i].die = TIME; @@ -666,10 +737,11 @@ void cluster_check_master(void) rebuild_address_pool(); // If we're not the very first master, this is a big issue! - if(count>0) + if (count > 0) LOG(0, 0, 0, "Warning: Fixed %d uninitialized sessions in becoming master!\n", count); config->cluster_undefined_sessions = 0; + config->cluster_undefined_bundles = 0; config->cluster_undefined_tunnels = 0; config->cluster_iam_uptodate = 1; // assume all peers are up-to-date @@ -687,7 +759,7 @@ void cluster_check_master(void) // we fix it up here, and we ensure that the 'first free session' // pointer is valid. // -static void cluster_check_sessions(int highsession, int freesession_ptr, int hightunnel) +static void cluster_check_sessions(int highsession, int freesession_ptr, int highbundle, int hightunnel) { int i; @@ -696,7 +768,7 @@ static void cluster_check_sessions(int highsession, int freesession_ptr, int hig if (config->cluster_iam_uptodate) return; - if (highsession > config->cluster_undefined_sessions && hightunnel > config->cluster_undefined_tunnels) + if (highsession > config->cluster_undefined_sessions && highbundle > config->cluster_undefined_bundles && hightunnel > config->cluster_undefined_tunnels) return; // Clear out defined sessions, counting the number of @@ -712,6 +784,19 @@ static void cluster_check_sessions(int highsession, int freesession_ptr, int hig ++config->cluster_undefined_sessions; } + // Clear out defined bundles, counting the number of + // undefs remaining. + config->cluster_undefined_bundles = 0; + for (i = 1 ; i < MAXBUNDLE; ++i) { + if (i > highbundle) { + if (bundle[i].state == BUNDLEUNDEF) bundle[i].state = BUNDLEFREE; // Defined. + continue; + } + + if (bundle[i].state == BUNDLEUNDEF) + ++config->cluster_undefined_bundles; + } + // Clear out defined tunnels, counting the number of // undefs remaining. config->cluster_undefined_tunnels = 0; @@ -726,9 +811,9 @@ static void cluster_check_sessions(int highsession, int freesession_ptr, int hig } - if (config->cluster_undefined_sessions || config->cluster_undefined_tunnels) { - LOG(2, 0, 0, "Cleared undefined sessions/tunnels. %d sess (high %d), %d tunn (high %d)\n", - config->cluster_undefined_sessions, highsession, config->cluster_undefined_tunnels, hightunnel); + if (config->cluster_undefined_sessions || config->cluster_undefined_tunnels || config->cluster_undefined_bundles) { + LOG(2, 0, 0, "Cleared undefined sessions/bundles/tunnels. %d sess (high %d), %d bund (high %d), %d tunn (high %d)\n", + config->cluster_undefined_sessions, highsession, config->cluster_undefined_bundles, highbundle, config->cluster_undefined_tunnels, hightunnel); return; } @@ -738,7 +823,7 @@ static void cluster_check_sessions(int highsession, int freesession_ptr, int hig cluster_uptodate(); } -static int hb_add_type(char **p, int type, int id) +static int hb_add_type(uint8_t **p, int type, int id) { switch (type) { case C_CSESSION: { // Compressed C_SESSION. @@ -752,13 +837,34 @@ static int hb_add_type(char **p, int type, int id) // Did we compress the full structure, and is the size actually // reduced?? if ( (d - orig) == sizeof(sessiont) && size < sizeof(sessiont) ) { - add_type(p, C_CSESSION, id, (char *) c, size); + add_type(p, C_CSESSION, id, c, size); break; } // Failed to compress : Fall through. } - case C_SESSION: add_type(p, C_SESSION, id, - (char *) &session[id], sizeof(sessiont)); + case C_SESSION: + add_type(p, C_SESSION, id, (uint8_t *) &session[id], sizeof(sessiont)); + break; + + case C_CBUNDLE: { // Compressed C_BUNDLE + uint8_t c[sizeof(bundlet) * 2]; // Bigger than worst case. + uint8_t *d = (uint8_t *) &bundle[id]; + uint8_t *orig = d; + int size; + + size = rle_compress( &d, sizeof(bundlet), c, sizeof(c) ); + + // Did we compress the full structure, and is the size actually + // reduced?? + if ( (d - orig) == sizeof(bundlet) && size < sizeof(bundlet) ) { + add_type(p, C_CBUNDLE, id, c, size); + break; + } + // Failed to compress : Fall through. + } + + case C_BUNDLE: + add_type(p, C_BUNDLE, id, (uint8_t *) &bundle[id], sizeof(bundlet)); break; case C_CTUNNEL: { // Compressed C_TUNNEL @@ -777,8 +883,8 @@ static int hb_add_type(char **p, int type, int id) } // Failed to compress : Fall through. } - case C_TUNNEL: add_type(p, C_TUNNEL, id, - (char *) &tunnel[id], sizeof(tunnelt)); + case C_TUNNEL: + add_type(p, C_TUNNEL, id, (uint8_t *) &tunnel[id], sizeof(tunnelt)); break; default: LOG(0, 0, 0, "Found an invalid type in heart queue! (%d)\n", type); @@ -793,10 +899,10 @@ static int hb_add_type(char **p, int type, int id) // void cluster_heartbeat() { - int i, count = 0, tcount = 0; - char buff[MAX_HEART_SIZE + sizeof(heartt) + sizeof(int) ]; + int i, count = 0, tcount = 0, bcount = 0; + uint8_t buff[MAX_HEART_SIZE + sizeof(heartt) + sizeof(int) ]; heartt h; - char *p = buff; + uint8_t *p = buff; if (!config->cluster_iam_master) // Only the master does this. return; @@ -814,13 +920,15 @@ void cluster_heartbeat() h.highsession = config->cluster_highest_sessionid; h.freesession = sessionfree; h.hightunnel = config->cluster_highest_tunnelid; + h.highbundle = config->cluster_highest_bundleid; h.size_sess = sizeof(sessiont); // Just in case. + h.size_bund = sizeof(bundlet); h.size_tunn = sizeof(tunnelt); h.interval = config->cluster_hb_interval; h.timeout = config->cluster_hb_timeout; h.table_version = config->cluster_table_version; - add_type(&p, C_HEARTBEAT, HB_VERSION, (char *) &h, sizeof(h)); + add_type(&p, C_HEARTBEAT, HB_VERSION, (uint8_t *) &h, sizeof(h)); for (i = 0; i < config->cluster_num_changes; ++i) { hb_add_type(&p, cluster_changes[i].type, cluster_changes[i].id); @@ -869,6 +977,21 @@ void cluster_heartbeat() ++tcount; } + // + // Fill out the packet with bundles from the bundle table... + while ( (p + sizeof(uint32_t) * 2 + sizeof(bundlet) ) < (buff + MAX_HEART_SIZE) ) { + + if (!walk_bundle_number) // bundle #0 isn't valid. + ++walk_bundle_number; + + if (bcount >= config->cluster_highest_bundleid) + break; + + hb_add_type(&p, C_CBUNDLE, walk_bundle_number); + walk_bundle_number = (1+walk_bundle_number)%(config->cluster_highest_bundleid+1); // +1 avoids divide by zero. + ++bcount; + } + // // Did we do something wrong? if (p > (buff + sizeof(buff))) { // Did we somehow manage to overun the buffer? @@ -877,10 +1000,10 @@ void cluster_heartbeat() exit(1); } - LOG(3, 0, 0, "Sending v%d heartbeat #%d, change #%" PRIu64 " with %d changes " - "(%d x-sess, %d x-tunnels, %d highsess, %d hightun, size %d)\n", + LOG(4, 0, 0, "Sending v%d heartbeat #%d, change #%" PRIu64 " with %d changes " + "(%d x-sess, %d x-bundles, %d x-tunnels, %d highsess, %d highbund, %d hightun, size %d)\n", HB_VERSION, h.seq, h.table_version, config->cluster_num_changes, - count, tcount, config->cluster_highest_sessionid, + count, bcount, tcount, config->cluster_highest_sessionid, config->cluster_highest_bundleid, config->cluster_highest_tunnelid, (int) (p - buff)); config->cluster_num_changes = 0; @@ -898,12 +1021,20 @@ static int type_changed(int type, int id) int i; for (i = 0 ; i < config->cluster_num_changes ; ++i) - if ( cluster_changes[i].id == id && - cluster_changes[i].type == type) - return 0; // Already marked for change. + { + if ( cluster_changes[i].id == id && cluster_changes[i].type == type) + { + // Already marked for change, remove it + --config->cluster_num_changes; + memmove(&cluster_changes[i], + &cluster_changes[i+1], + (config->cluster_num_changes - i) * sizeof(cluster_changes[i])); + break; + } + } - cluster_changes[i].type = type; - cluster_changes[i].id = id; + cluster_changes[config->cluster_num_changes].type = type; + cluster_changes[config->cluster_num_changes].id = id; ++config->cluster_num_changes; if (config->cluster_num_changes > MAX_CHANGES) @@ -912,7 +1043,6 @@ static int type_changed(int type, int id) return 1; } - // A particular session has been changed! int cluster_send_session(int sid) { @@ -929,6 +1059,17 @@ int cluster_send_session(int sid) return type_changed(C_CSESSION, sid); } +// A particular bundle has been changed! +int cluster_send_bundle(int bid) +{ + if (!config->cluster_iam_master) { + LOG(0, 0, bid, "I'm not a master, but I just tried to change a bundle!\n"); + return -1; + } + + return type_changed(C_CBUNDLE, bid); +} + // A particular tunnel has been changed! int cluster_send_tunnel(int tid) { @@ -1099,7 +1240,7 @@ static int cluster_set_master(in_addr_t peer, in_addr_t master) // Note that we don't mark the session as dirty; We rely on // the slow table walk to propogate this back out to the slaves. // -static int cluster_handle_bytes(char *data, int size) +static int cluster_handle_bytes(uint8_t *data, int size) { bytest *b; @@ -1126,7 +1267,9 @@ static int cluster_handle_bytes(char *data, int size) session[b->sid].cout_delta += b->cout; if (b->cin) - session[b->sid].last_packet = time_now; // Reset idle timer! + session[b->sid].last_packet = session[b->sid].last_data = time_now; + else if (b->cout) + session[b->sid].last_data = time_now; size -= sizeof(*b); ++b; @@ -1166,6 +1309,31 @@ static int cluster_recv_session(int more, uint8_t *p) return 0; } +static int cluster_recv_bundle(int more, uint8_t *p) +{ + if (more >= MAXBUNDLE) { + LOG(0, 0, 0, "DANGER: Received a bundle id > MAXBUNDLE!\n"); + return -1; + } + + if (bundle[more].state == BUNDLEUNDEF) { + if (config->cluster_iam_uptodate) { // Sanity. + LOG(0, 0, 0, "I thought I was uptodate but I just found an undefined bundle!\n"); + } else { + --config->cluster_undefined_bundles; + } + } + + memcpy(&bundle[more], p, sizeof(bundle[more]) ); + + LOG(5, 0, more, "Received bundle update\n"); + + if (!config->cluster_iam_uptodate) + cluster_uptodate(); // Check to see if we're up to date. + + return 0; +} + static int cluster_recv_tunnel(int more, uint8_t *p) { if (more >= MAXTUNNEL) { @@ -1199,51 +1367,184 @@ static int cluster_recv_tunnel(int more, uint8_t *p) } -// pre v5 heartbeat session structure +// pre v6 heartbeat session structure struct oldsession { sessionidt next; sessionidt far; tunnelidt tunnel; + uint8_t flags; + struct { + uint8_t phase; + uint8_t lcp:4; + uint8_t ipcp:4; + uint8_t ipv6cp:4; + uint8_t ccp:4; + } ppp; + char reserved_1[2]; in_addr_t ip; int ip_pool_index; - unsigned long unique_id; - uint16_t nr; - uint16_t ns; + uint32_t unique_id; + char reserved_2[4]; uint32_t magic; - uint32_t cin, cout; uint32_t pin, pout; - uint32_t total_cin; - uint32_t total_cout; - uint32_t id; + uint32_t cin, cout; + uint32_t cin_wrap, cout_wrap; + uint32_t cin_delta, cout_delta; uint16_t throttle_in; uint16_t throttle_out; + uint8_t filter_in; + uint8_t filter_out; + uint16_t mru; clockt opened; clockt die; + uint32_t session_timeout; + uint32_t idle_timeout; time_t last_packet; + time_t last_data; in_addr_t dns1, dns2; routet route[MAXROUTE]; - uint16_t radius; - uint16_t mru; uint16_t tbf_in; uint16_t tbf_out; - uint8_t l2tp_flags; - uint8_t reserved_old_snoop; - uint8_t walled_garden; - uint8_t flags1; - char random_vector[MAXTEL]; int random_vector_length; - char user[129]; + uint8_t random_vector[MAXTEL]; + char user[MAXUSER]; char called[MAXTEL]; char calling[MAXTEL]; uint32_t tx_connect_speed; uint32_t rx_connect_speed; - uint32_t flags; + clockt timeout; + uint32_t mrru; + uint8_t mssf; + epdist epdis; + bundleidt bundle; in_addr_t snoop_ip; uint16_t snoop_port; - uint16_t sid; - uint8_t filter_in; - uint8_t filter_out; - char reserved[18]; + uint8_t walled_garden; + uint8_t ipv6prefixlen; + struct in6_addr ipv6route; + char reserved_3[11]; +}; + +struct oldsessionV7 { + sessionidt next; // next session in linked list + sessionidt far; // far end session ID + tunnelidt tunnel; // near end tunnel ID + uint8_t flags; // session flags: see SESSION_* + struct { + uint8_t phase; // PPP phase + uint8_t lcp:4; // LCP state + uint8_t ipcp:4; // IPCP state + uint8_t ipv6cp:4; // IPV6CP state + uint8_t ccp:4; // CCP state + } ppp; + uint16_t mru; // maximum receive unit + in_addr_t ip; // IP of session set by RADIUS response (host byte order). + int ip_pool_index; // index to IP pool + uint32_t unique_id; // unique session id + uint32_t magic; // ppp magic number + uint32_t pin, pout; // packet counts + uint32_t cin, cout; // byte counts + uint32_t cin_wrap, cout_wrap; // byte counter wrap count (RADIUS accounting giagawords) + uint32_t cin_delta, cout_delta; // byte count changes (for dump_session()) + uint16_t throttle_in; // upstream throttle rate (kbps) + uint16_t throttle_out; // downstream throttle rate + uint8_t filter_in; // input filter index (to ip_filters[N-1]; 0 if none) + uint8_t filter_out; // output filter index + uint16_t snoop_port; // Interception destination port + in_addr_t snoop_ip; // Interception destination IP + clockt opened; // when started + clockt die; // being closed, when to finally free + uint32_t session_timeout; // Maximum session time in seconds + uint32_t idle_timeout; // Maximum idle time in seconds + time_t last_packet; // Last packet from the user (used for idle timeouts) + time_t last_data; // Last data packet to/from the user (used for idle timeouts) + in_addr_t dns1, dns2; // DNS servers + routet route[MAXROUTE]; // static routes + uint16_t tbf_in; // filter bucket for throttling in from the user. + uint16_t tbf_out; // filter bucket for throttling out to the user. + int random_vector_length; + uint8_t random_vector[MAXTEL]; + char user[MAXUSER]; // user (needed in session for radius stop messages) + char called[MAXTEL]; // called number + char calling[MAXTEL]; // calling number + uint32_t tx_connect_speed; + uint32_t rx_connect_speed; + clockt timeout; // Session timeout + uint32_t mrru; // Multilink Max-Receive-Reconstructed-Unit + epdist epdis; // Multilink Endpoint Discriminator + bundleidt bundle; // Multilink Bundle Identifier + uint8_t mssf; // Multilink Short Sequence Number Header Format + uint8_t walled_garden; // is this session gardened? + uint8_t classlen; // class (needed for radius accounting messages) + char class[MAXCLASS]; + uint8_t ipv6prefixlen; // IPv6 route prefix length + struct in6_addr ipv6route; // Static IPv6 route + sessionidt forwardtosession; // LNS id_session to forward + uint8_t src_hwaddr[ETH_ALEN]; // MAC addr source (for pppoe sessions 6 bytes) + char reserved[4]; // Space to expand structure without changing HB_VERSION +}; + +struct oldsessionV8 { + sessionidt next; // next session in linked list + sessionidt far; // far end session ID + tunnelidt tunnel; // near end tunnel ID + uint8_t flags; // session flags: see SESSION_* + struct { + uint8_t phase; // PPP phase + uint8_t lcp:4; // LCP state + uint8_t ipcp:4; // IPCP state + uint8_t ipv6cp:4; // IPV6CP state + uint8_t ccp:4; // CCP state + } ppp; + uint16_t mru; // maximum receive unit + in_addr_t ip; // IP of session set by RADIUS response (host byte order). + int ip_pool_index; // index to IP pool + uint32_t unique_id; // unique session id + uint32_t magic; // ppp magic number + uint32_t pin, pout; // packet counts + uint32_t cin, cout; // byte counts + uint32_t cin_wrap, cout_wrap; // byte counter wrap count (RADIUS accounting giagawords) + uint32_t cin_delta, cout_delta; // byte count changes (for dump_session()) + uint16_t throttle_in; // upstream throttle rate (kbps) + uint16_t throttle_out; // downstream throttle rate + uint8_t filter_in; // input filter index (to ip_filters[N-1]; 0 if none) + uint8_t filter_out; // output filter index + uint16_t snoop_port; // Interception destination port + in_addr_t snoop_ip; // Interception destination IP + clockt opened; // when started + clockt die; // being closed, when to finally free + uint32_t session_timeout; // Maximum session time in seconds + uint32_t idle_timeout; // Maximum idle time in seconds + time_t last_packet; // Last packet from the user (used for idle timeouts) + time_t last_data; // Last data packet to/from the user (used for idle timeouts) + in_addr_t dns1, dns2; // DNS servers + routet route[MAXROUTE]; // static routes + uint16_t tbf_in; // filter bucket for throttling in from the user. + uint16_t tbf_out; // filter bucket for throttling out to the user. + int random_vector_length; + uint8_t random_vector[MAXTEL]; + char user[MAXUSER]; // user (needed in session for radius stop messages) + char called[MAXTEL]; // called number + char calling[MAXTEL]; // calling number + uint32_t tx_connect_speed; + uint32_t rx_connect_speed; + clockt timeout; // Session timeout + uint32_t mrru; // Multilink Max-Receive-Reconstructed-Unit + epdist epdis; // Multilink Endpoint Discriminator + bundleidt bundle; // Multilink Bundle Identifier + uint8_t mssf; // Multilink Short Sequence Number Header Format + uint8_t walled_garden; // is this session gardened? + uint8_t classlen; // class (needed for radius accounting messages) + char class[MAXCLASS]; + uint8_t ipv6prefixlen; // IPv6 route prefix length + struct in6_addr ipv6route; // Static IPv6 route + sessionidt forwardtosession; // LNS id_session to forward + uint8_t src_hwaddr[ETH_ALEN]; // MAC addr source (for pppoe sessions 6 bytes) + uint32_t dhcpv6_prefix_iaid; // prefix iaid requested by client + uint32_t dhcpv6_iana_iaid; // iaid of iana requested by client + struct in6_addr ipv6address; // Framed Ipv6 address + struct dhcp6_opt_clientid dhcpv6_client_id; // Size max (headers + DUID) + char reserved[4]; // Space to expand structure without changing HB_VERSION }; static uint8_t *convert_session(struct oldsession *old) @@ -1256,20 +1557,24 @@ static uint8_t *convert_session(struct oldsession *old) new.next = old->next; new.far = old->far; new.tunnel = old->tunnel; - new.l2tp_flags = old->l2tp_flags; new.flags = old->flags; + new.ppp.phase = old->ppp.phase; + new.ppp.lcp = old->ppp.lcp; + new.ppp.ipcp = old->ppp.ipcp; + new.ppp.ipv6cp = old->ppp.ipv6cp; + new.ppp.ccp = old->ppp.ccp; new.ip = old->ip; new.ip_pool_index = old->ip_pool_index; new.unique_id = old->unique_id; - new.nr = old->nr; - new.ns = old->ns; new.magic = old->magic; new.pin = old->pin; new.pout = old->pout; - new.cin = old->total_cin; - new.cout = old->total_cout; - new.cin_delta = old->cin; - new.cout_delta = old->cout; + new.cin = old->cin; + new.cout = old->cout; + new.cin_wrap = old->cin_wrap; + new.cout_wrap = old->cout_wrap; + new.cin_delta = old->cin_delta; + new.cout_delta = old->cout_delta; new.throttle_in = old->throttle_in; new.throttle_out = old->throttle_out; new.filter_in = old->filter_in; @@ -1277,7 +1582,10 @@ static uint8_t *convert_session(struct oldsession *old) new.mru = old->mru; new.opened = old->opened; new.die = old->die; + new.session_timeout = old->session_timeout; + new.idle_timeout = old->idle_timeout; new.last_packet = old->last_packet; + new.last_data = old->last_data; new.dns1 = old->dns1; new.dns2 = old->dns2; new.tbf_in = old->tbf_in; @@ -1285,9 +1593,16 @@ static uint8_t *convert_session(struct oldsession *old) new.random_vector_length = old->random_vector_length; new.tx_connect_speed = old->tx_connect_speed; new.rx_connect_speed = old->rx_connect_speed; + new.timeout = old->timeout; + new.mrru = old->mrru; + new.mssf = old->mssf; + new.epdis = old->epdis; + new.bundle = old->bundle; new.snoop_ip = old->snoop_ip; new.snoop_port = old->snoop_port; new.walled_garden = old->walled_garden; + new.route6[0].ipv6prefixlen = old->ipv6prefixlen; + new.route6[0].ipv6route = old->ipv6route; memcpy(new.random_vector, old->random_vector, sizeof(new.random_vector)); memcpy(new.user, old->user, sizeof(new.user)); @@ -1300,12 +1615,157 @@ static uint8_t *convert_session(struct oldsession *old) return (uint8_t *) &new; } +static uint8_t *convert_sessionV7(struct oldsessionV7 *old) +{ + static sessiont new; + int i; + + memset(&new, 0, sizeof(new)); + + new.next = old->next; + new.far = old->far; + new.tunnel = old->tunnel; + new.flags = old->flags; + new.ppp.phase = old->ppp.phase; + new.ppp.lcp = old->ppp.lcp; + new.ppp.ipcp = old->ppp.ipcp; + new.ppp.ipv6cp = old->ppp.ipv6cp; + new.ppp.ccp = old->ppp.ccp; + new.mru = old->mru; + new.ip = old->ip; + new.ip_pool_index = old->ip_pool_index; + new.unique_id = old->unique_id; + new.magic = old->magic; + new.pin = old->pin; + new.pout = old->pout; + new.cin = old->cin; + new.cout = old->cout; + new.cin_wrap = old->cin_wrap; + new.cout_wrap = old->cout_wrap; + new.cin_delta = old->cin_delta; + new.cout_delta = old->cout_delta; + new.throttle_in = old->throttle_in; + new.throttle_out = old->throttle_out; + new.filter_in = old->filter_in; + new.filter_out = old->filter_out; + new.snoop_port = old->snoop_port; + new.snoop_ip = old->snoop_ip; + new.opened = old->opened; + new.die = old->die; + new.session_timeout = old->session_timeout; + new.idle_timeout = old->idle_timeout; + new.last_packet = old->last_packet; + new.last_data = old->last_data; + new.dns1 = old->dns1; + new.dns2 = old->dns2; + for (i = 0; i < MAXROUTE; i++) + memcpy(&new.route[i], &old->route[i], sizeof(new.route[i])); + new.tbf_in = old->tbf_in; + new.tbf_out = old->tbf_out; + new.random_vector_length = old->random_vector_length; + memcpy(new.random_vector, old->random_vector, sizeof(new.random_vector)); + memcpy(new.user, old->user, sizeof(new.user)); + memcpy(new.called, old->called, sizeof(new.called)); + memcpy(new.calling, old->calling, sizeof(new.calling)); + new.tx_connect_speed = old->tx_connect_speed; + new.rx_connect_speed = old->rx_connect_speed; + new.timeout = old->timeout; + new.mrru = old->mrru; + new.epdis = old->epdis; + new.bundle = old->bundle; + new.mssf = old->mssf; + new.walled_garden = old->walled_garden; + new.classlen = old->classlen; + memcpy(new.class, old->class, sizeof(new.class)); + new.route6[0].ipv6prefixlen = old->ipv6prefixlen; + new.route6[0].ipv6route = old->ipv6route; + + new.forwardtosession = old->forwardtosession; + memcpy(new.src_hwaddr, old->src_hwaddr, sizeof(new.src_hwaddr)); + + return (uint8_t *) &new; +} + +static uint8_t *convert_sessionV8(struct oldsessionV8 *old) +{ + static sessiont new; + int i; + + memset(&new, 0, sizeof(new)); + + new.next = old->next; + new.far = old->far; + new.tunnel = old->tunnel; + new.flags = old->flags; + new.ppp.phase = old->ppp.phase; + new.ppp.lcp = old->ppp.lcp; + new.ppp.ipcp = old->ppp.ipcp; + new.ppp.ipv6cp = old->ppp.ipv6cp; + new.ppp.ccp = old->ppp.ccp; + new.mru = old->mru; + new.ip = old->ip; + new.ip_pool_index = old->ip_pool_index; + new.unique_id = old->unique_id; + new.magic = old->magic; + new.pin = old->pin; + new.pout = old->pout; + new.cin = old->cin; + new.cout = old->cout; + new.cin_wrap = old->cin_wrap; + new.cout_wrap = old->cout_wrap; + new.cin_delta = old->cin_delta; + new.cout_delta = old->cout_delta; + new.throttle_in = old->throttle_in; + new.throttle_out = old->throttle_out; + new.filter_in = old->filter_in; + new.filter_out = old->filter_out; + new.snoop_port = old->snoop_port; + new.snoop_ip = old->snoop_ip; + new.opened = old->opened; + new.die = old->die; + new.session_timeout = old->session_timeout; + new.idle_timeout = old->idle_timeout; + new.last_packet = old->last_packet; + new.last_data = old->last_data; + new.dns1 = old->dns1; + new.dns2 = old->dns2; + for (i = 0; i < MAXROUTE; i++) + memcpy(&new.route[i], &old->route[i], sizeof(new.route[i])); + new.tbf_in = old->tbf_in; + new.tbf_out = old->tbf_out; + new.random_vector_length = old->random_vector_length; + memcpy(new.random_vector, old->random_vector, sizeof(new.random_vector)); + memcpy(new.user, old->user, sizeof(new.user)); + memcpy(new.called, old->called, sizeof(new.called)); + memcpy(new.calling, old->calling, sizeof(new.calling)); + new.tx_connect_speed = old->tx_connect_speed; + new.rx_connect_speed = old->rx_connect_speed; + new.timeout = old->timeout; + new.mrru = old->mrru; + new.epdis = old->epdis; + new.bundle = old->bundle; + new.mssf = old->mssf; + new.walled_garden = old->walled_garden; + new.classlen = old->classlen; + memcpy(new.class, old->class, sizeof(new.class)); + new.route6[0].ipv6prefixlen = old->ipv6prefixlen; + new.route6[0].ipv6route = old->ipv6route; + + new.forwardtosession = old->forwardtosession; + memcpy(new.src_hwaddr, old->src_hwaddr, sizeof(new.src_hwaddr)); + new.dhcpv6_prefix_iaid = old->dhcpv6_prefix_iaid; + new.dhcpv6_iana_iaid = old->dhcpv6_iana_iaid; + new.ipv6address = old->ipv6address; + new.dhcpv6_client_id = old->dhcpv6_client_id; + + return (uint8_t *) &new; +} + // // Process a heartbeat.. // -// v3: added interval, timeout -// v4: added table_version -// v5: added ipv6, re-ordered session structure +// v6: added RADIUS class attribute, re-ordered session structure +// v7: added tunnelt attribute at the end of struct (tunnelt size change) static int cluster_process_heartbeat(uint8_t *data, int size, int more, uint8_t *p, in_addr_t addr) { heartt *h; @@ -1313,17 +1773,21 @@ static int cluster_process_heartbeat(uint8_t *data, int size, int more, uint8_t int i, type; int hb_ver = more; -#if HB_VERSION != 5 +#if HB_VERSION != 9 # error "need to update cluster_process_heartbeat()" #endif - // we handle versions 3 through 5 - if (hb_ver < 3 || hb_ver > HB_VERSION) { + // we handle versions 5 through 8 + if (hb_ver < 5 || hb_ver > HB_VERSION) { LOG(0, 0, 0, "Received a heartbeat version that I don't support (%d)!\n", hb_ver); return -1; // Ignore it?? } - // Ok. It's a heartbeat packet from a cluster master! + if (size > sizeof(past_hearts[0].data)) { + LOG(0, 0, 0, "Received an oversize heartbeat from %s (%d)!\n", fmtaddr(addr, 0), size); + return -1; + } + if (s < sizeof(*h)) goto shortpacket; @@ -1343,18 +1807,17 @@ static int cluster_process_heartbeat(uint8_t *data, int size, int more, uint8_t return -1; // Skip it. } - if (hb_ver >= 4) { - if (h->table_version > config->cluster_table_version) { - LOG(0, 0, 0, "They've seen more state changes (%" PRIu64 " vs my %" PRIu64 ") so I'm gone!\n", + if (h->table_version > config->cluster_table_version) { + LOG(0, 0, 0, "They've seen more state changes (%" PRIu64 " vs my %" PRIu64 ") so I'm gone!\n", h->table_version, config->cluster_table_version); - kill(0, SIGTERM); - exit(1); - } - if (h->table_version < config->cluster_table_version) - return -1; + kill(0, SIGTERM); + exit(1); } + if (h->table_version < config->cluster_table_version) + return -1; + if (basetime > h->basetime) { LOG(0, 0, 0, "They're an older master than me so I'm gone!\n"); kill(0, SIGTERM); @@ -1437,7 +1900,7 @@ static int cluster_process_heartbeat(uint8_t *data, int size, int more, uint8_t // Check that we don't have too many undefined sessions, and // that the free session pointer is correct. - cluster_check_sessions(h->highsession, h->freesession, h->hightunnel); + cluster_check_sessions(h->highsession, h->freesession, h->highbundle, h->hightunnel); if (h->interval != config->cluster_hb_interval) { @@ -1476,7 +1939,7 @@ static int cluster_process_heartbeat(uint8_t *data, int size, int more, uint8_t s -= (p - orig_p); // session struct changed with v5 - if (hb_ver < 5) + if (hb_ver < 6) { if (size != sizeof(struct oldsession)) { LOG(0, 0, 0, "DANGER: Received a v%d CSESSION that didn't decompress correctly!\n", hb_ver); @@ -1487,17 +1950,30 @@ static int cluster_process_heartbeat(uint8_t *data, int size, int more, uint8_t break; } - if (size != sizeof(sessiont) ) { // Ouch! Very very bad! - LOG(0, 0, 0, "DANGER: Received a CSESSION that didn't decompress correctly!\n"); - // Now what? Should exit! No-longer up to date! - break; + if (size != sizeof(sessiont)) { // Ouch! Very very bad! + if ((hb_ver < HB_VERSION) && (size < sizeof(sessiont))) + { + if ((hb_ver == 7) && (size == sizeof(struct oldsessionV7))) + cluster_recv_session(more, convert_sessionV7((struct oldsessionV7 *) c)); + else if (size == sizeof(struct oldsessionV8)) + cluster_recv_session(more, convert_sessionV8((struct oldsessionV8 *) c)); + else + LOG(0, 0, 0, "DANGER: Received a CSESSION version=%d that didn't decompress correctly!\n", hb_ver); + break; + } + else + { + LOG(0, 0, 0, "DANGER: Received a CSESSION that didn't decompress correctly!\n"); + // Now what? Should exit! No-longer up to date! + break; + } } cluster_recv_session(more, c); break; } case C_SESSION: - if (hb_ver < 5) + if (hb_ver < 6) { if (s < sizeof(struct oldsession)) goto shortpacket; @@ -1526,8 +2002,10 @@ static int cluster_process_heartbeat(uint8_t *data, int size, int more, uint8_t size = rle_decompress((uint8_t **) &p, s, c, sizeof(c)); s -= (p - orig_p); - if (size != sizeof(tunnelt) ) { // Ouch! Very very bad! - LOG(0, 0, 0, "DANGER: Received a CSESSION that didn't decompress correctly!\n"); + if ( ((hb_ver >= HB_VERSION) && (size != sizeof(tunnelt))) || + ((hb_ver < HB_VERSION) && (size > sizeof(tunnelt))) ) + { // Ouch! Very very bad! + LOG(0, 0, 0, "DANGER: Received a CTUNNEL that didn't decompress correctly!\n"); // Now what? Should exit! No-longer up to date! break; } @@ -1545,6 +2023,34 @@ static int cluster_process_heartbeat(uint8_t *data, int size, int more, uint8_t p += sizeof(tunnel[more]); s -= sizeof(tunnel[more]); break; + + case C_CBUNDLE: { // Compressed bundle structure. + uint8_t c[ sizeof(bundlet) + 2]; + int size; + uint8_t *orig_p = p; + + size = rle_decompress((uint8_t **) &p, s, c, sizeof(c)); + s -= (p - orig_p); + + if (size != sizeof(bundlet) ) { // Ouch! Very very bad! + LOG(0, 0, 0, "DANGER: Received a CBUNDLE that didn't decompress correctly!\n"); + // Now what? Should exit! No-longer up to date! + break; + } + + cluster_recv_bundle(more, c); + break; + + } + case C_BUNDLE: + if ( s < sizeof(bundle[more])) + goto shortpacket; + + cluster_recv_bundle(more, p); + + p += sizeof(bundle[more]); + s -= sizeof(bundle[more]); + break; default: LOG(0, 0, 0, "DANGER: I received a heartbeat element where I didn't understand the type! (%d)\n", type); return -1; // can't process any more of the packet!! @@ -1572,10 +2078,10 @@ shortpacket: // We got a packet on the cluster port! // Handle pings, lastseens, and heartbeats! // -int processcluster(char *data, int size, in_addr_t addr) +int processcluster(uint8_t *data, int size, in_addr_t addr) { int type, more; - char *p = data; + uint8_t *p = data; int s = size; if (addr == my_address) @@ -1612,7 +2118,7 @@ int processcluster(char *data, int size, in_addr_t addr) case C_FORWARD_DAE: // Forwarded DAE packet. pass off to processdae. if (!config->cluster_iam_master) { - LOG(0, 0, 0, "I'm not the master, but I got a C_FORWARD_%s from %s?\n", + LOG(0, 0, 0, "I'm not the master, but I got a C_FORWARD%s from %s?\n", type == C_FORWARD_DAE ? "_DAE" : "", fmtaddr(addr, 0)); return -1; @@ -1620,9 +2126,11 @@ int processcluster(char *data, int size, in_addr_t addr) else { struct sockaddr_in a; + uint16_t indexudp; a.sin_addr.s_addr = more; - a.sin_port = *(int *) p; + a.sin_port = (*(int *) p) & 0xFFFF; + indexudp = ((*(int *) p) >> 16) & 0xFFFF; s -= sizeof(int); p += sizeof(int); @@ -1631,12 +2139,37 @@ int processcluster(char *data, int size, in_addr_t addr) STAT(recv_forward); if (type == C_FORWARD_DAE) - processdae(p, s, &a, sizeof(a)); + { + struct in_addr local; + local.s_addr = config->bind_address ? config->bind_address : my_address; + processdae(p, s, &a, sizeof(a), &local); + } else - processudp(p, s, &a); + processudp(p, s, &a, indexudp); return 0; } + case C_PPPOE_FORWARD: + if (!config->cluster_iam_master) + { + LOG(0, 0, 0, "I'm not the master, but I got a C_PPPOE_FORWARD from %s?\n", fmtaddr(addr, 0)); + return -1; + } + else + { + pppoe_process_forward(p, s, addr); + return 0; + } + + case C_MPPP_FORWARD: + // Receive a MPPP packet from a slave. + if (!config->cluster_iam_master) { + LOG(0, 0, 0, "I'm not the master, but I got a C_MPPP_FORWARD from %s?\n", fmtaddr(addr, 0)); + return -1; + } + + processipout(p, s); + return 0; case C_THROTTLE: { // Receive a forwarded packet from a slave. if (!config->cluster_iam_master) { @@ -1703,7 +2236,7 @@ shortpacket: //==================================================================================================== -int cmd_show_cluster(struct cli_def *cli, char *command, char **argv, int argc) +int cmd_show_cluster(struct cli_def *cli, const char *command, char **argv, int argc) { int i; @@ -1714,6 +2247,7 @@ int cmd_show_cluster(struct cli_def *cli, char *command, char **argv, int argc) cli_print(cli, "My address : %s", fmtaddr(my_address, 0)); cli_print(cli, "VIP address : %s", fmtaddr(config->bind_address, 0)); cli_print(cli, "Multicast address: %s", fmtaddr(config->cluster_address, 0)); + cli_print(cli, "UDP port : %u", config->cluster_port); cli_print(cli, "Multicast i'face : %s", config->cluster_interface); if (!config->cluster_iam_master) { @@ -1726,11 +2260,13 @@ int cmd_show_cluster(struct cli_def *cli, char *command, char **argv, int argc) cli_print(cli, "Table version # : %" PRIu64, config->cluster_table_version); cli_print(cli, "Next sequence number expected: %d", config->cluster_seq_number); cli_print(cli, "%d sessions undefined of %d", config->cluster_undefined_sessions, config->cluster_highest_sessionid); + cli_print(cli, "%d bundles undefined of %d", config->cluster_undefined_bundles, config->cluster_highest_bundleid); cli_print(cli, "%d tunnels undefined of %d", config->cluster_undefined_tunnels, config->cluster_highest_tunnelid); } else { cli_print(cli, "Table version # : %" PRIu64, config->cluster_table_version); cli_print(cli, "Next heartbeat # : %d", config->cluster_seq_number); cli_print(cli, "Highest session : %d", config->cluster_highest_sessionid); + cli_print(cli, "Highest bundle : %d", config->cluster_highest_bundleid); cli_print(cli, "Highest tunnel : %d", config->cluster_highest_tunnelid); cli_print(cli, "%d changes queued for sending", config->cluster_num_changes); } @@ -1814,7 +2350,7 @@ static int rle_decompress(uint8_t **src_p, int ssize, uint8_t *dst, int dsize) { int count; int orig_dsize = dsize; - char *src = *src_p; + uint8_t *src = *src_p; while (ssize >0 && dsize > 0) { // While there's more to decompress, and there's room in the decompress buffer... count = *src++; --ssize; // get the count byte from the source.