conntrackd: generalize/cleanup network message building/parsing

This patch generalizes the network message building and parsing
to prepare the upcoming expectation support.

Basically, it renames:

- NET_T_STATE_* by NET_T_STATE_CT_*, as I plan to add NET_T_STATE_EXP_*
- BUILD_NETMSG by BUILD_NETMSG_FROM_CT, and build_payload by ct2msg.
  I plan to add exp2msg.
- parse_payload by msg2ct, since I plan to add msg2exp.
- modify object_status_to_network_type to prepare the support of
  expectations.
- add prefix ct_ to all parsing functions in parse.c, as we will have
  similar functions to convert messages to expectation objects.

Signed-off-by: Pablo Neira Ayuso <pablo@netfilter.org>

11 files changed, 155 insertions, 131 deletions

diff --git a/include/cache.h b/include/cache.h
index a42e395..02bb386 100644
--- a/include/cache.h
+++ b/include/cache.h
@@ -21,7 +21,8 @@ enum {
 	C_OBJ_NONE = 0,		/* not in the cache */
 	C_OBJ_NEW,		/* just added to the cache */
 	C_OBJ_ALIVE,		/* in the cache, alive */
-	C_OBJ_DEAD		/* still in the cache, but dead */
+	C_OBJ_DEAD,		/* still in the cache, but dead */
+	C_OBJ_MAX
 };
 
 struct cache;
diff --git a/include/network.h b/include/network.h
index 567317b..d0531b9 100644
--- a/include/network.h
+++ b/include/network.h
@@ -25,10 +25,10 @@ struct nethdr {
 #define NETHDR_SIZ nethdr_align(sizeof(struct nethdr))
 
 enum nethdr_type {
-	NET_T_STATE_NEW = 0,
-	NET_T_STATE_UPD,
-	NET_T_STATE_DEL,
-	NET_T_STATE_MAX = NET_T_STATE_DEL,
+	NET_T_STATE_CT_NEW = 0,
+	NET_T_STATE_CT_UPD,
+	NET_T_STATE_CT_DEL,
+	NET_T_STATE_MAX = NET_T_STATE_CT_DEL,
 	NET_T_CTL = 10,
 };
 
@@ -37,7 +37,9 @@ int nethdr_size(int len);
 void nethdr_set(struct nethdr *net, int type);
 void nethdr_set_ack(struct nethdr *net);
 void nethdr_set_ctl(struct nethdr *net);
-int object_status_to_network_type(int status);
+
+struct cache_object;
+int object_status_to_network_type(struct cache_object *obj);
 
 #define NETHDR_DATA(x)							 \
 	(struct netattr *)(((char *)x) + NETHDR_SIZ)
@@ -79,13 +81,13 @@ enum {
 	MSG_BAD,
 };
 
-#define BUILD_NETMSG(ct, query)					\
+#define BUILD_NETMSG_FROM_CT(ct, query)				\
 ({								\
 	static char __net[4096];				\
 	struct nethdr *__hdr = (struct nethdr *) __net;		\
 	memset(__hdr, 0, NETHDR_SIZ);				\
 	nethdr_set(__hdr, query);				\
-	build_payload(ct, __hdr);				\
+	ct2msg(ct, __hdr);					\
 	HDR_HOST2NETWORK(__hdr);				\
 	__hdr;							\
 })
@@ -234,8 +236,7 @@ struct nta_attr_natseqadj {
 	uint32_t repl_seq_offset_after;
 };
 
-void build_payload(const struct nf_conntrack *ct, struct nethdr *n);
-
-int parse_payload(struct nf_conntrack *ct, struct nethdr *n, size_t remain);
+void ct2msg(const struct nf_conntrack *ct, struct nethdr *n);
+int msg2ct(struct nf_conntrack *ct, struct nethdr *n, size_t remain);
 
 #endif
diff --git a/src/build.c b/src/build.c
index a495872..9c3687c 100644
--- a/src/build.c
+++ b/src/build.c
@@ -1,6 +1,7 @@
 /*
- * (C) 2006-2008 by Pablo Neira Ayuso <pablo@netfilter.org>
- * 
+ * (C) 2006-2011 by Pablo Neira Ayuso <pablo@netfilter.org>
+ * (C) 2011 by Vyatta Inc. <http://www.vyatta.com>
+ *
  * This program is free software; you can redistribute it and/or modify
  * it under the terms of the GNU General Public License as published by
  * the Free Software Foundation; either version 2 of the License, or
@@ -42,14 +43,14 @@ addattr(struct nethdr *n, int attr, const void *data, size_t len)
 }
 
 static inline void
-__build_u8(const struct nf_conntrack *ct, int a, struct nethdr *n, int b)
+ct_build_u8(const struct nf_conntrack *ct, int a, struct nethdr *n, int b)
 {
 	void *ptr = put_header(n, b, sizeof(uint8_t));
 	memcpy(ptr, nfct_get_attr(ct, a), sizeof(uint8_t));
 }
 
 static inline void 
-__build_u16(const struct nf_conntrack *ct, int a, struct nethdr *n, int b)
+ct_build_u16(const struct nf_conntrack *ct, int a, struct nethdr *n, int b)
 {
 	uint16_t data = nfct_get_attr_u16(ct, a);
 	data = htons(data);
@@ -57,7 +58,7 @@ __build_u16(const struct nf_conntrack *ct, int a, struct nethdr *n, int b)
 }
 
 static inline void 
-__build_u32(const struct nf_conntrack *ct, int a, struct nethdr *n, int b)
+ct_build_u32(const struct nf_conntrack *ct, int a, struct nethdr *n, int b)
 {
 	uint32_t data = nfct_get_attr_u32(ct, a);
 	data = htonl(data);
@@ -65,7 +66,7 @@ __build_u32(const struct nf_conntrack *ct, int a, struct nethdr *n, int b)
 }
 
 static inline void 
-__build_group(const struct nf_conntrack *ct, int a, struct nethdr *n, 
+ct_build_group(const struct nf_conntrack *ct, int a, struct nethdr *n, 
 	      int b, int size)
 {
 	void *ptr = put_header(n, b, size);
@@ -73,7 +74,7 @@ __build_group(const struct nf_conntrack *ct, int a, struct nethdr *n,
 }
 
 static inline void 
-__build_natseqadj(const struct nf_conntrack *ct, struct nethdr *n)
+ct_build_natseqadj(const struct nf_conntrack *ct, struct nethdr *n)
 {
 	struct nta_attr_natseqadj data = {
 		.orig_seq_correction_pos =
@@ -99,54 +100,54 @@ static enum nf_conntrack_attr nat_type[] =
 
 static void build_l4proto_tcp(const struct nf_conntrack *ct, struct nethdr *n)
 {
-	__build_group(ct, ATTR_GRP_ORIG_PORT, n, NTA_PORT,
+	ct_build_group(ct, ATTR_GRP_ORIG_PORT, n, NTA_PORT,
 		      sizeof(struct nfct_attr_grp_port));
 
 	if (!nfct_attr_is_set(ct, ATTR_TCP_STATE))
 		return;
 
-	__build_u8(ct, ATTR_TCP_STATE, n, NTA_TCP_STATE);
+	ct_build_u8(ct, ATTR_TCP_STATE, n, NTA_TCP_STATE);
 	if (CONFIG(sync).tcp_window_tracking) {
-		__build_u8(ct, ATTR_TCP_WSCALE_ORIG, n, NTA_TCP_WSCALE_ORIG);
-		__build_u8(ct, ATTR_TCP_WSCALE_REPL, n, NTA_TCP_WSCALE_REPL);
+		ct_build_u8(ct, ATTR_TCP_WSCALE_ORIG, n, NTA_TCP_WSCALE_ORIG);
+		ct_build_u8(ct, ATTR_TCP_WSCALE_REPL, n, NTA_TCP_WSCALE_REPL);
 	}
 }
 
 static void build_l4proto_sctp(const struct nf_conntrack *ct, struct nethdr *n)
 {
-	__build_group(ct, ATTR_GRP_ORIG_PORT, n, NTA_PORT,
+	ct_build_group(ct, ATTR_GRP_ORIG_PORT, n, NTA_PORT,
 		      sizeof(struct nfct_attr_grp_port));
 
 	if (!nfct_attr_is_set(ct, ATTR_SCTP_STATE))
 		return;
 
-	__build_u8(ct, ATTR_SCTP_STATE, n, NTA_SCTP_STATE);
-	__build_u32(ct, ATTR_SCTP_VTAG_ORIG, n, NTA_SCTP_VTAG_ORIG);
-	__build_u32(ct, ATTR_SCTP_VTAG_REPL, n, NTA_SCTP_VTAG_REPL);
+	ct_build_u8(ct, ATTR_SCTP_STATE, n, NTA_SCTP_STATE);
+	ct_build_u32(ct, ATTR_SCTP_VTAG_ORIG, n, NTA_SCTP_VTAG_ORIG);
+	ct_build_u32(ct, ATTR_SCTP_VTAG_REPL, n, NTA_SCTP_VTAG_REPL);
 }
 
 static void build_l4proto_dccp(const struct nf_conntrack *ct, struct nethdr *n)
 {
-	__build_group(ct, ATTR_GRP_ORIG_PORT, n, NTA_PORT,
+	ct_build_group(ct, ATTR_GRP_ORIG_PORT, n, NTA_PORT,
 		      sizeof(struct nfct_attr_grp_port));
 
 	if (!nfct_attr_is_set(ct, ATTR_DCCP_STATE))
 		return;
 
-	__build_u8(ct, ATTR_DCCP_STATE, n, NTA_DCCP_STATE);
-	__build_u8(ct, ATTR_DCCP_ROLE, n, NTA_DCCP_ROLE);
+	ct_build_u8(ct, ATTR_DCCP_STATE, n, NTA_DCCP_STATE);
+	ct_build_u8(ct, ATTR_DCCP_ROLE, n, NTA_DCCP_ROLE);
 }
 
 static void build_l4proto_icmp(const struct nf_conntrack *ct, struct nethdr *n)
 {
-	__build_u8(ct, ATTR_ICMP_TYPE, n, NTA_ICMP_TYPE);
-	__build_u8(ct, ATTR_ICMP_CODE, n, NTA_ICMP_CODE);
-	__build_u16(ct, ATTR_ICMP_ID, n, NTA_ICMP_ID);
+	ct_build_u8(ct, ATTR_ICMP_TYPE, n, NTA_ICMP_TYPE);
+	ct_build_u8(ct, ATTR_ICMP_CODE, n, NTA_ICMP_CODE);
+	ct_build_u16(ct, ATTR_ICMP_ID, n, NTA_ICMP_ID);
 }
 
 static void build_l4proto_udp(const struct nf_conntrack *ct, struct nethdr *n)
 {
-	__build_group(ct, ATTR_GRP_ORIG_PORT, n, NTA_PORT,
+	ct_build_group(ct, ATTR_GRP_ORIG_PORT, n, NTA_PORT,
 		      sizeof(struct nfct_attr_grp_port));
 }
 
@@ -165,45 +166,45 @@ static struct build_l4proto {
 	[IPPROTO_UDP]		= { .build = build_l4proto_udp },
 };
 
-void build_payload(const struct nf_conntrack *ct, struct nethdr *n)
+void ct2msg(const struct nf_conntrack *ct, struct nethdr *n)
 {
 	uint8_t l4proto = nfct_get_attr_u8(ct, ATTR_L4PROTO);
 
 	if (nfct_attr_grp_is_set(ct, ATTR_GRP_ORIG_IPV4)) {
-		__build_group(ct, ATTR_GRP_ORIG_IPV4, n, NTA_IPV4, 
+		ct_build_group(ct, ATTR_GRP_ORIG_IPV4, n, NTA_IPV4, 
 			      sizeof(struct nfct_attr_grp_ipv4));
 	} else if (nfct_attr_grp_is_set(ct, ATTR_GRP_ORIG_IPV6)) {
-		__build_group(ct, ATTR_GRP_ORIG_IPV6, n, NTA_IPV6, 
+		ct_build_group(ct, ATTR_GRP_ORIG_IPV6, n, NTA_IPV6, 
 			      sizeof(struct nfct_attr_grp_ipv6));
 	}
 
-	__build_u32(ct, ATTR_STATUS, n, NTA_STATUS); 
-	__build_u8(ct, ATTR_L4PROTO, n, NTA_L4PROTO);
+	ct_build_u32(ct, ATTR_STATUS, n, NTA_STATUS); 
+	ct_build_u8(ct, ATTR_L4PROTO, n, NTA_L4PROTO);
 
 	if (l4proto_fcn[l4proto].build)
 		l4proto_fcn[l4proto].build(ct, n);
 
 	if (!CONFIG(commit_timeout) && nfct_attr_is_set(ct, ATTR_TIMEOUT))
-		__build_u32(ct, ATTR_TIMEOUT, n, NTA_TIMEOUT);
+		ct_build_u32(ct, ATTR_TIMEOUT, n, NTA_TIMEOUT);
 	if (nfct_attr_is_set(ct, ATTR_MARK))
-		__build_u32(ct, ATTR_MARK, n, NTA_MARK);
+		ct_build_u32(ct, ATTR_MARK, n, NTA_MARK);
 
 	/* setup the master conntrack */
 	if (nfct_attr_grp_is_set(ct, ATTR_GRP_MASTER_IPV4)) {
-		__build_group(ct, ATTR_GRP_MASTER_IPV4, n, NTA_MASTER_IPV4,
+		ct_build_group(ct, ATTR_GRP_MASTER_IPV4, n, NTA_MASTER_IPV4,
 			      sizeof(struct nfct_attr_grp_ipv4));
-		__build_u8(ct, ATTR_MASTER_L4PROTO, n, NTA_MASTER_L4PROTO);
+		ct_build_u8(ct, ATTR_MASTER_L4PROTO, n, NTA_MASTER_L4PROTO);
 		if (nfct_attr_grp_is_set(ct, ATTR_GRP_MASTER_PORT)) {
-			__build_group(ct, ATTR_GRP_MASTER_PORT,
+			ct_build_group(ct, ATTR_GRP_MASTER_PORT,
 				      n, NTA_MASTER_PORT, 
 				      sizeof(struct nfct_attr_grp_port));
 		}
 	} else if (nfct_attr_grp_is_set(ct, ATTR_GRP_MASTER_IPV6)) {
-		__build_group(ct, ATTR_GRP_MASTER_IPV6, n, NTA_MASTER_IPV6,
+		ct_build_group(ct, ATTR_GRP_MASTER_IPV6, n, NTA_MASTER_IPV6,
 			      sizeof(struct nfct_attr_grp_ipv6));
-		__build_u8(ct, ATTR_MASTER_L4PROTO, n, NTA_MASTER_L4PROTO);
+		ct_build_u8(ct, ATTR_MASTER_L4PROTO, n, NTA_MASTER_L4PROTO);
 		if (nfct_attr_grp_is_set(ct, ATTR_GRP_MASTER_PORT)) {
-			__build_group(ct, ATTR_GRP_MASTER_PORT,
+			ct_build_group(ct, ATTR_GRP_MASTER_PORT,
 				      n, NTA_MASTER_PORT,
 				      sizeof(struct nfct_attr_grp_port));
 		}
@@ -211,15 +212,15 @@ void build_payload(const struct nf_conntrack *ct, struct nethdr *n)
 
 	/*  NAT */
 	if (nfct_getobjopt(ct, NFCT_GOPT_IS_SNAT))
-		__build_u32(ct, ATTR_REPL_IPV4_DST, n, NTA_SNAT_IPV4);
+		ct_build_u32(ct, ATTR_REPL_IPV4_DST, n, NTA_SNAT_IPV4);
 	if (nfct_getobjopt(ct, NFCT_GOPT_IS_DNAT))
-		__build_u32(ct, ATTR_REPL_IPV4_SRC, n, NTA_DNAT_IPV4);
+		ct_build_u32(ct, ATTR_REPL_IPV4_SRC, n, NTA_DNAT_IPV4);
 	if (nfct_getobjopt(ct, NFCT_GOPT_IS_SPAT))
-		__build_u16(ct, ATTR_REPL_PORT_DST, n, NTA_SPAT_PORT);
+		ct_build_u16(ct, ATTR_REPL_PORT_DST, n, NTA_SPAT_PORT);
 	if (nfct_getobjopt(ct, NFCT_GOPT_IS_DPAT))
-		__build_u16(ct, ATTR_REPL_PORT_SRC, n, NTA_DPAT_PORT);
+		ct_build_u16(ct, ATTR_REPL_PORT_SRC, n, NTA_DPAT_PORT);
 
 	/* NAT sequence adjustment */
 	if (nfct_attr_is_set_array(ct, nat_type, 6))
-		__build_natseqadj(ct, n);
+		ct_build_natseqadj(ct, n);
 }
diff --git a/src/internal_bypass.c b/src/internal_bypass.c
index 8ecec34..98717f3 100644
--- a/src/internal_bypass.c
+++ b/src/internal_bypass.c
@@ -118,7 +118,7 @@ static void internal_bypass_ct_event_new(struct nf_conntrack *ct, int origin)
 	if (origin != CTD_ORIGIN_NOT_ME)
 		return;
 
-	net = BUILD_NETMSG(ct, NET_T_STATE_NEW);
+	net = BUILD_NETMSG_FROM_CT(ct, NET_T_STATE_CT_NEW);
 	multichannel_send(STATE_SYNC(channel), net);
 	internal_bypass_stats.new++;
 }
@@ -131,7 +131,7 @@ static void internal_bypass_ct_event_upd(struct nf_conntrack *ct, int origin)
 	if (origin != CTD_ORIGIN_NOT_ME)
 		return;
 
-	net = BUILD_NETMSG(ct, NET_T_STATE_UPD);
+	net = BUILD_NETMSG_FROM_CT(ct, NET_T_STATE_CT_UPD);
 	multichannel_send(STATE_SYNC(channel), net);
 	internal_bypass_stats.upd++;
 }
@@ -144,7 +144,7 @@ static int internal_bypass_ct_event_del(struct nf_conntrack *ct, int origin)
 	if (origin != CTD_ORIGIN_NOT_ME)
 		return 1;
 
-	net = BUILD_NETMSG(ct, NET_T_STATE_DEL);
+	net = BUILD_NETMSG_FROM_CT(ct, NET_T_STATE_CT_DEL);
 	multichannel_send(STATE_SYNC(channel), net);
 	internal_bypass_stats.del++;
 
diff --git a/src/internal_cache.c b/src/internal_cache.c
index 7a698e6..952327d 100644
--- a/src/internal_cache.c
+++ b/src/internal_cache.c
@@ -81,7 +81,7 @@ static int internal_cache_ct_purge_step(void *data1, void *data2)
 	if (!STATE(get_retval)) {
 		if (obj->status != C_OBJ_DEAD) {
 			cache_object_set_status(obj, C_OBJ_DEAD);
-			sync_send(obj, NET_T_STATE_DEL);
+			sync_send(obj, NET_T_STATE_CT_DEL);
 			cache_object_put(obj);
 		}
 	}
@@ -117,10 +117,10 @@ internal_cache_ct_resync(enum nf_conntrack_msg_type type,
 
 	switch (obj->status) {
 	case C_OBJ_NEW:
-		sync_send(obj, NET_T_STATE_NEW);
+		sync_send(obj, NET_T_STATE_CT_NEW);
 		break;
 	case C_OBJ_ALIVE:
-		sync_send(obj, NET_T_STATE_UPD);
+		sync_send(obj, NET_T_STATE_CT_UPD);
 		break;
 	}
 	return NFCT_CB_CONTINUE;
@@ -155,7 +155,7 @@ retry:
 		 * processes or the kernel, but don't propagate events that
 		 * have been triggered by conntrackd itself, eg. commits. */
 		if (origin == CTD_ORIGIN_NOT_ME)
-			sync_send(obj, NET_T_STATE_NEW);
+			sync_send(obj, NET_T_STATE_CT_NEW);
 	} else {
 		cache_del(STATE(mode)->internal->ct.data, obj);
 		cache_object_free(obj);
@@ -176,7 +176,7 @@ static void internal_cache_ct_event_upd(struct nf_conntrack *ct, int origin)
 		return;
 
 	if (origin == CTD_ORIGIN_NOT_ME)
-		sync_send(obj, NET_T_STATE_UPD);
+		sync_send(obj, NET_T_STATE_CT_UPD);
 }
 
 static int internal_cache_ct_event_del(struct nf_conntrack *ct, int origin)
@@ -196,7 +196,7 @@ static int internal_cache_ct_event_del(struct nf_conntrack *ct, int origin)
 	if (obj->status != C_OBJ_DEAD) {
 		cache_object_set_status(obj, C_OBJ_DEAD);
 		if (origin == CTD_ORIGIN_NOT_ME) {
-			sync_send(obj, NET_T_STATE_DEL);
+			sync_send(obj, NET_T_STATE_CT_DEL);
 		}
 		cache_object_put(obj);
 	}
diff --git a/src/network.c b/src/network.c
index 6a66a2b..cadc466 100644
--- a/src/network.c
+++ b/src/network.c
@@ -1,6 +1,7 @@
 /*
- * (C) 2006-2007 by Pablo Neira Ayuso <pablo@netfilter.org>
- * 
+ * (C) 2006-2011 by Pablo Neira Ayuso <pablo@netfilter.org>
+ * (C) 2011 by Vyatta Inc. <http://www.vyatta.com>
+ *
  * This program is free software; you can redistribute it and/or modify
  * it under the terms of the GNU General Public License as published by
  * the Free Software Foundation; either version 2 of the License, or
@@ -119,13 +120,15 @@ int nethdr_track_is_seq_set()
 
 #include "cache.h"
 
-static int status2type[] = {
-	[C_OBJ_NEW]	= NET_T_STATE_NEW,
-	[C_OBJ_ALIVE]	= NET_T_STATE_UPD,
-	[C_OBJ_DEAD]	= NET_T_STATE_DEL,
+static int status2type[CACHE_T_MAX][C_OBJ_MAX] = {
+	[CACHE_T_CT] = {
+		[C_OBJ_NEW]	= NET_T_STATE_CT_NEW,
+		[C_OBJ_ALIVE]	= NET_T_STATE_CT_UPD,
+		[C_OBJ_DEAD]	= NET_T_STATE_CT_DEL,
+	},
 };
 
-int object_status_to_network_type(int status)
+int object_status_to_network_type(struct cache_object *obj)
 {
-	return status2type[status];
+	return status2type[obj->cache->type][obj->status];
 }
diff --git a/src/parse.c b/src/parse.c
index 7e60597..0718128 100644
--- a/src/parse.c
+++ b/src/parse.c
@@ -1,6 +1,7 @@
 /*
- * (C) 2006-2007 by Pablo Neira Ayuso <pablo@netfilter.org>
- * 
+ * (C) 2006-2011 by Pablo Neira Ayuso <pablo@netfilter.org>
+ * (C) 2011 by Vyatta Inc. <http://www.vyatta.com>
+ *
  * This program is free software; you can redistribute it and/or modify
  * it under the terms of the GNU General Public License as published by
  * the Free Software Foundation; either version 2 of the License, or
@@ -24,184 +25,184 @@
 #define ssizeof(x) (int)sizeof(x)
 #endif
 
-static void parse_u8(struct nf_conntrack *ct, int attr, void *data);
-static void parse_u16(struct nf_conntrack *ct, int attr, void *data);
-static void parse_u32(struct nf_conntrack *ct, int attr, void *data);
-static void parse_group(struct nf_conntrack *ct, int attr, void *data);
-static void parse_nat_seq_adj(struct nf_conntrack *ct, int attr, void *data);
+static void ct_parse_u8(struct nf_conntrack *ct, int attr, void *data);
+static void ct_parse_u16(struct nf_conntrack *ct, int attr, void *data);
+static void ct_parse_u32(struct nf_conntrack *ct, int attr, void *data);
+static void ct_parse_group(struct nf_conntrack *ct, int attr, void *data);
+static void ct_parse_nat_seq_adj(struct nf_conntrack *ct, int attr, void *data);
 
-struct parser {
+struct ct_parser {
 	void 	(*parse)(struct nf_conntrack *ct, int attr, void *data);
 	int 	attr;
 	int	size;
 };
 
-static struct parser h[NTA_MAX] = {
+static struct ct_parser h[NTA_MAX] = {
 	[NTA_IPV4] = {
-		.parse	= parse_group,
+		.parse	= ct_parse_group,
 		.attr	= ATTR_GRP_ORIG_IPV4,
 		.size	= NTA_SIZE(sizeof(struct nfct_attr_grp_ipv4)),
 	},
 	[NTA_IPV6] = {
-		.parse	= parse_group,
+		.parse	= ct_parse_group,
 		.attr	= ATTR_GRP_ORIG_IPV6,
 		.size	= NTA_SIZE(sizeof(struct nfct_attr_grp_ipv6)),
 	},
 	[NTA_PORT] = {
-		.parse	= parse_group,
+		.parse	= ct_parse_group,
 		.attr	= ATTR_GRP_ORIG_PORT,
 		.size	= NTA_SIZE(sizeof(struct nfct_attr_grp_port)),
 	},
 	[NTA_L4PROTO] = {
-		.parse	= parse_u8,
+		.parse	= ct_parse_u8,
 		.attr	= ATTR_L4PROTO,
 		.size	= NTA_SIZE(sizeof(uint8_t)),
 	},
 	[NTA_TCP_STATE] = {
-		.parse	= parse_u8,
+		.parse	= ct_parse_u8,
 		.attr	= ATTR_TCP_STATE,
 		.size	= NTA_SIZE(sizeof(uint8_t)),
 	},
 	[NTA_STATUS] = {
-		.parse	= parse_u32,
+		.parse	= ct_parse_u32,
 		.attr	= ATTR_STATUS,
 		.size	= NTA_SIZE(sizeof(uint32_t)),
 	},
 	[NTA_MARK] = {
-		.parse	= parse_u32,
+		.parse	= ct_parse_u32,
 		.attr	= ATTR_MARK,
 		.size	= NTA_SIZE(sizeof(uint32_t)),
 	},
 	[NTA_TIMEOUT] = {
-		.parse	= parse_u32,
+		.parse	= ct_parse_u32,
 		.attr	= ATTR_TIMEOUT,
 		.size	= NTA_SIZE(sizeof(uint32_t)),
 	},
 	[NTA_MASTER_IPV4] = {
-		.parse	= parse_group,
+		.parse	= ct_parse_group,
 		.attr	= ATTR_GRP_MASTER_IPV4,
 		.size	= NTA_SIZE(sizeof(struct nfct_attr_grp_ipv4)),
 	},
 	[NTA_MASTER_IPV6] = {
-		.parse	= parse_group,
+		.parse	= ct_parse_group,
 		.attr	= ATTR_GRP_MASTER_IPV6,
 		.size	= NTA_SIZE(sizeof(struct nfct_attr_grp_ipv6)),
 	},
 	[NTA_MASTER_L4PROTO] = {
-		.parse	= parse_u8,
+		.parse	= ct_parse_u8,
 		.attr	= ATTR_MASTER_L4PROTO,
 		.size	= NTA_SIZE(sizeof(uint8_t)),
 	},
 	[NTA_MASTER_PORT] = {
-		.parse	= parse_group,
+		.parse	= ct_parse_group,
 		.attr	= ATTR_GRP_MASTER_PORT,
 		.size	= NTA_SIZE(sizeof(struct nfct_attr_grp_port)),
 	},
 	[NTA_SNAT_IPV4]	= {
-		.parse	= parse_u32,
+		.parse	= ct_parse_u32,
 		.attr	= ATTR_SNAT_IPV4,
 		.size	= NTA_SIZE(sizeof(uint32_t)),
 	},
 	[NTA_DNAT_IPV4] = {
-		.parse	= parse_u32,
+		.parse	= ct_parse_u32,
 		.attr	= ATTR_DNAT_IPV4,
 		.size	= NTA_SIZE(sizeof(uint32_t)),
 	},
 	[NTA_SPAT_PORT]	= {
-		.parse	= parse_u16,
+		.parse	= ct_parse_u16,
 		.attr	= ATTR_SNAT_PORT,
 		.size	= NTA_SIZE(sizeof(uint16_t)),
 	},
 	[NTA_DPAT_PORT]	= {
-		.parse	= parse_u16,
+		.parse	= ct_parse_u16,
 		.attr	= ATTR_DNAT_PORT,
 		.size	= NTA_SIZE(sizeof(uint16_t)),
 	},
 	[NTA_NAT_SEQ_ADJ] = {
-		.parse	= parse_nat_seq_adj,
+		.parse	= ct_parse_nat_seq_adj,
 		.size	= NTA_SIZE(sizeof(struct nta_attr_natseqadj)),
 	},
 	[NTA_SCTP_STATE] = {
-		.parse	= parse_u8,
+		.parse	= ct_parse_u8,
 		.attr	= ATTR_SCTP_STATE,
 		.size	= NTA_SIZE(sizeof(uint8_t)),
 	},
 	[NTA_SCTP_VTAG_ORIG] = {
-		.parse	= parse_u32,
+		.parse	= ct_parse_u32,
 		.attr	= ATTR_SCTP_VTAG_ORIG,
 		.size	= NTA_SIZE(sizeof(uint32_t)),
 	},
 	[NTA_SCTP_VTAG_REPL] = {
-		.parse	= parse_u32,
+		.parse	= ct_parse_u32,
 		.attr	= ATTR_SCTP_VTAG_REPL,
 		.size	= NTA_SIZE(sizeof(uint32_t)),
 	},
 	[NTA_DCCP_STATE] = {
-		.parse	= parse_u8,
+		.parse	= ct_parse_u8,
 		.attr	= ATTR_DCCP_STATE,
 		.size	= NTA_SIZE(sizeof(uint8_t)),
 	},
 	[NTA_DCCP_ROLE] = {
-		.parse	= parse_u8,
+		.parse	= ct_parse_u8,
 		.attr	= ATTR_DCCP_ROLE,
 		.size	= NTA_SIZE(sizeof(uint8_t)),
 	},
 	[NTA_ICMP_TYPE] = {
-		.parse	= parse_u8,
+		.parse	= ct_parse_u8,
 		.attr	= ATTR_ICMP_TYPE,
 		.size	= NTA_SIZE(sizeof(uint8_t)),
 	},
 	[NTA_ICMP_CODE] = {
-		.parse	= parse_u8,
+		.parse	= ct_parse_u8,
 		.attr	= ATTR_ICMP_CODE,
 		.size	= NTA_SIZE(sizeof(uint8_t)),
 	},
 	[NTA_ICMP_ID] = {
-		.parse	= parse_u16,
+		.parse	= ct_parse_u16,
 		.attr	= ATTR_ICMP_ID,
 		.size	= NTA_SIZE(sizeof(uint16_t)),
 	},
 	[NTA_TCP_WSCALE_ORIG] = {
-		.parse	= parse_u8,
+		.parse	= ct_parse_u8,
 		.attr	= ATTR_TCP_WSCALE_ORIG,
 		.size	= NTA_SIZE(sizeof(uint8_t)),
 	},
 	[NTA_TCP_WSCALE_REPL] = {
-		.parse	= parse_u8,
+		.parse	= ct_parse_u8,
 		.attr	= ATTR_TCP_WSCALE_REPL,
 		.size	= NTA_SIZE(sizeof(uint8_t)),
 	},
 };
 
 static void
-parse_u8(struct nf_conntrack *ct, int attr, void *data)
+ct_parse_u8(struct nf_conntrack *ct, int attr, void *data)
 {
 	uint8_t *value = (uint8_t *) data;
 	nfct_set_attr_u8(ct, h[attr].attr, *value);
 }
 
 static void
-parse_u16(struct nf_conntrack *ct, int attr, void *data)
+ct_parse_u16(struct nf_conntrack *ct, int attr, void *data)
 {
 	uint16_t *value = (uint16_t *) data;
 	nfct_set_attr_u16(ct, h[attr].attr, ntohs(*value));
 }
 
 static void
-parse_u32(struct nf_conntrack *ct, int attr, void *data)
+ct_parse_u32(struct nf_conntrack *ct, int attr, void *data)
 {
 	uint32_t *value = (uint32_t *) data;
 	nfct_set_attr_u32(ct, h[attr].attr, ntohl(*value));
 }
 
 static void
-parse_group(struct nf_conntrack *ct, int attr, void *data)
+ct_parse_group(struct nf_conntrack *ct, int attr, void *data)
 {
 	nfct_set_attr_grp(ct, h[attr].attr, data);
 }
 
 static void
-parse_nat_seq_adj(struct nf_conntrack *ct, int attr, void *data)
+ct_parse_nat_seq_adj(struct nf_conntrack *ct, int attr, void *data)
 {
 	struct nta_attr_natseqadj *this = data;
 	nfct_set_attr_u32(ct, ATTR_ORIG_NAT_SEQ_CORRECTION_POS, 
@@ -218,7 +219,7 @@ parse_nat_seq_adj(struct nf_conntrack *ct, int attr, void *data)
 			  ntohl(this->repl_seq_offset_after));
 }
 
-int parse_payload(struct nf_conntrack *ct, struct nethdr *net, size_t remain)
+int msg2ct(struct nf_conntrack *ct, struct nethdr *net, size_t remain)
 {
 	int len;
 	struct netattr *attr;
diff --git a/src/sync-alarm.c b/src/sync-alarm.c
index 8d6b34d..65154a1 100644
--- a/src/sync-alarm.c
+++ b/src/sync-alarm.c
@@ -42,7 +42,7 @@ static void refresher(struct alarm_block *a, void *data)
 		  random() % CONFIG(refresh) + 1,
 		  ((random() % 5 + 1)  * 200000) - 1);
 
-	alarm_enqueue(obj, NET_T_STATE_UPD);
+	alarm_enqueue(obj, NET_T_STATE_CT_UPD);
 }
 
 static void cache_alarm_add(struct cache_object *obj, void *data)
@@ -137,7 +137,7 @@ static int tx_queue_xmit(struct queue_node *n, const void *data)
 
 		ca = (struct cache_alarm *)n;
 		obj = cache_data_get_object(STATE(mode)->internal->ct.data, ca);
-		type = object_status_to_network_type(obj->status);
+		type = object_status_to_network_type(obj);
 		net = obj->cache->ops->build_msg(obj, type);
 		multichannel_send(STATE_SYNC(channel), net);
 		cache_object_put(obj);
diff --git a/src/sync-ftfw.c b/src/sync-ftfw.c
index 55eda0b..cff4d25 100644
--- a/src/sync-ftfw.c
+++ b/src/sync-ftfw.c
@@ -518,7 +518,7 @@ static int tx_queue_xmit(struct queue_node *n, const void *data)
 
 		cn = (struct cache_ftfw *)n;
 		obj = cache_data_get_object(STATE(mode)->internal->ct.data, cn);
-		type = object_status_to_network_type(obj->status);
+		type = object_status_to_network_type(obj);
 		net = obj->cache->ops->build_msg(obj, type);
 		nethdr_set_hello(net);
 
diff --git a/src/sync-mode.c b/src/sync-mode.c
index 7f019f7..17533f8 100644
--- a/src/sync-mode.c
+++ b/src/sync-mode.c
@@ -41,6 +41,24 @@
 #include <net/if.h>
 #include <fcntl.h>
 
+static struct nf_conntrack *msg2ct_alloc(struct nethdr *net, size_t remain)
+{
+	struct nf_conntrack *ct;
+
+	/* TODO: add stats on ENOMEM errors in the future. */
+	ct = nfct_new();
+	if (ct == NULL)
+		return NULL;
+
+	if (msg2ct(ct, net, remain) == -1) {
+		STATE_SYNC(error).msg_rcv_malformed++;
+		STATE_SYNC(error).msg_rcv_bad_payload++;
+		nfct_destroy(ct);
+		return NULL;
+	}
+	return ct;
+}
+
 static void
 do_channel_handler_step(int i, struct nethdr *net, size_t remain)
 {
@@ -74,26 +92,24 @@ do_channel_handler_step(int i, struct nethdr *net, size_t remain)
 		STATE_SYNC(error).msg_rcv_bad_type++;
 		return;
 	}
-	/* TODO: add stats on ENOMEM errors in the future. */
-	ct = nfct_new();
-	if (ct == NULL)
-		return;
-
-	if (parse_payload(ct, net, remain) == -1) {
-		STATE_SYNC(error).msg_rcv_malformed++;
-		STATE_SYNC(error).msg_rcv_bad_payload++;
-		nfct_destroy(ct);
-		return;
-	}
 
 	switch(net->type) {
-	case NET_T_STATE_NEW:
+	case NET_T_STATE_CT_NEW:
+		ct = msg2ct_alloc(net, remain);
+		if (ct == NULL)
+			return;
 		STATE_SYNC(external)->ct.new(ct);
 		break;
-	case NET_T_STATE_UPD:
+	case NET_T_STATE_CT_UPD:
+		ct = msg2ct_alloc(net, remain);
+		if (ct == NULL)
+			return;
 		STATE_SYNC(external)->ct.upd(ct);
 		break;
-	case NET_T_STATE_DEL:
+	case NET_T_STATE_CT_DEL:
+		ct = msg2ct_alloc(net, remain);
+		if (ct == NULL)
+			return;
 		STATE_SYNC(external)->ct.del(ct);
 		break;
 	default:
diff --git a/src/sync-notrack.c b/src/sync-notrack.c
index e25cfd8..6c798ac 100644
--- a/src/sync-notrack.c
+++ b/src/sync-notrack.c
@@ -1,6 +1,7 @@
 /*
- * (C) 2008 by Pablo Neira Ayuso <pablo@netfilter.org>
- * 
+ * (C) 2006-2011 by Pablo Neira Ayuso <pablo@netfilter.org>
+ * (C) 2011 by Vyatta Inc. <http://www.vyatta.com>
+ *
  * This program is free software; you can redistribute it and/or modify
  * it under the terms of the GNU General Public License as published by
  * the Free Software Foundation; either version 2 of the License, or
@@ -87,7 +88,7 @@ static int kernel_resync_cb(enum nf_conntrack_msg_type type,
 {
 	struct nethdr *net;
 
-	net = BUILD_NETMSG(ct, NET_T_STATE_NEW);
+	net = BUILD_NETMSG_FROM_CT(ct, NET_T_STATE_CT_NEW);
 	multichannel_send(STATE_SYNC(channel), net);
 
 	return NFCT_CB_CONTINUE;
@@ -198,7 +199,7 @@ static int tx_queue_xmit(struct queue_node *n, const void *data2)
 
 		cn = (struct cache_ftfw *)n;
 		obj = cache_data_get_object(STATE(mode)->internal->ct.data, cn);
-		type = object_status_to_network_type(obj->status);;
+		type = object_status_to_network_type(obj);;
 		net = obj->cache->ops->build_msg(obj, type);
 
 		multichannel_send(STATE_SYNC(channel), net);