tipc-discussion — User questions & problem reports, project news, developer discussion

Re: [tipc-discussion] [PATCH net 2/2] tipc: set ub->ifindex for local ipv6 address

[Ying X. <yin...@wi...>]

On 8/3/20 11:34 PM, Xin Long wrote:
> Without ub->ifindex set for ipv6 address in tipc_udp_enable(),
> ipv6_sock_mc_join() may make the wrong dev join the multicast
> address in enable_mcast(). This causes that tipc links would
> never be created.
> 
> So fix it by getting the right netdev and setting ub->ifindex,
> as it does for ipv4 address.
> 
> Reported-by: Shuang Li <sh...@re...>
> Signed-off-by: Xin Long <luc...@gm...>

Acked-by: Ying Xue <yin...@wi...>

> ---
>  net/tipc/udp_media.c | 8 ++++++++
>  1 file changed, 8 insertions(+)
> 
> diff --git a/net/tipc/udp_media.c b/net/tipc/udp_media.c
> index 28a283f..9dec596 100644
> --- a/net/tipc/udp_media.c
> +++ b/net/tipc/udp_media.c
> @@ -738,6 +738,13 @@ static int tipc_udp_enable(struct net *net, struct tipc_bearer *b,
>  		b->mtu = b->media->mtu;
>  #if IS_ENABLED(CONFIG_IPV6)
>  	} else if (local.proto == htons(ETH_P_IPV6)) {
> +		struct net_device *dev;
> +
> +		dev = ipv6_dev_find(net, &local.ipv6);
> +		if (!dev) {
> +			err = -ENODEV;
> +			goto err;
> +		}
>  		udp_conf.family = AF_INET6;
>  		udp_conf.use_udp6_tx_checksums = true;
>  		udp_conf.use_udp6_rx_checksums = true;
> @@ -745,6 +752,7 @@ static int tipc_udp_enable(struct net *net, struct tipc_bearer *b,
>  			udp_conf.local_ip6 = in6addr_any;
>  		else
>  			udp_conf.local_ip6 = local.ipv6;
> +		ub->ifindex = dev->ifindex;
>  		b->mtu = 1280;
>  #endif
>  	} else {
> 

Re: [tipc-discussion] [PATCH net 1/2] ipv6: add ipv6_dev_find()

[Ying X. <yin...@wi...>]

On 8/3/20 11:34 PM, Xin Long wrote:
> This is to add an ip_dev_find like function for ipv6, used to find
> the dev by saddr.
> 
> It will be used by TIPC protocol. So also export it.
> 
> Signed-off-by: Xin Long <luc...@gm...>

Acked-by: Ying Xue <yin...@wi...>

> ---
>  include/net/addrconf.h |  2 ++
>  net/ipv6/addrconf.c    | 39 +++++++++++++++++++++++++++++++++++++++
>  2 files changed, 41 insertions(+)
> 
> diff --git a/include/net/addrconf.h b/include/net/addrconf.h
> index 8418b7d..ba3f6c15 100644
> --- a/include/net/addrconf.h
> +++ b/include/net/addrconf.h
> @@ -97,6 +97,8 @@ bool ipv6_chk_custom_prefix(const struct in6_addr *addr,
>  
>  int ipv6_chk_prefix(const struct in6_addr *addr, struct net_device *dev);
>  
> +struct net_device *ipv6_dev_find(struct net *net, const struct in6_addr *addr);
> +
>  struct inet6_ifaddr *ipv6_get_ifaddr(struct net *net,
>  				     const struct in6_addr *addr,
>  				     struct net_device *dev, int strict);
> diff --git a/net/ipv6/addrconf.c b/net/ipv6/addrconf.c
> index 840bfdb..857d6f9 100644
> --- a/net/ipv6/addrconf.c
> +++ b/net/ipv6/addrconf.c
> @@ -1983,6 +1983,45 @@ int ipv6_chk_prefix(const struct in6_addr *addr, struct net_device *dev)
>  }
>  EXPORT_SYMBOL(ipv6_chk_prefix);
>  
> +/**
> + * ipv6_dev_find - find the first device with a given source address.
> + * @net: the net namespace
> + * @addr: the source address
> + *
> + * The caller should be protected by RCU, or RTNL.
> + */
> +struct net_device *ipv6_dev_find(struct net *net, const struct in6_addr *addr)
> +{
> +	unsigned int hash = inet6_addr_hash(net, addr);
> +	struct inet6_ifaddr *ifp, *result = NULL;
> +	struct net_device *dev = NULL;
> +
> +	rcu_read_lock();
> +	hlist_for_each_entry_rcu(ifp, &inet6_addr_lst[hash], addr_lst) {
> +		if (net_eq(dev_net(ifp->idev->dev), net) &&
> +		    ipv6_addr_equal(&ifp->addr, addr)) {
> +			result = ifp;
> +			break;
> +		}
> +	}
> +
> +	if (!result) {
> +		struct rt6_info *rt;
> +
> +		rt = rt6_lookup(net, addr, NULL, 0, NULL, 0);
> +		if (rt) {
> +			dev = rt->dst.dev;
> +			ip6_rt_put(rt);
> +		}
> +	} else {
> +		dev = result->idev->dev;
> +	}
> +	rcu_read_unlock();
> +
> +	return dev;
> +}
> +EXPORT_SYMBOL(ipv6_dev_find);
> +
>  struct inet6_ifaddr *ipv6_get_ifaddr(struct net *net, const struct in6_addr *addr,
>  				     struct net_device *dev, int strict)
>  {
> 

Re: [tipc-discussion] [PATCH net-next] tipc: Use is_broadcast_ether_addr() instead of memcmp()

[David M. <da...@da...>]

From: Huang Guobin <hua...@hu...>
Date: Sun, 2 Aug 2020 22:00:55 -0400

> Using is_broadcast_ether_addr() instead of directly use
> memcmp() to determine if the ethernet address is broadcast
> address.
> 
> spatch with a semantic match is used to found this problem.
> (http://coccinelle.lip6.fr/)
> 
> Signed-off-by: Huang Guobin <hua...@hu...>

Applied, thank you.

[tipc-discussion] [PATCH net 2/2] tipc: set ub->ifindex for local ipv6 address

[Xin L. <luc...@gm...>]

Without ub->ifindex set for ipv6 address in tipc_udp_enable(),
ipv6_sock_mc_join() may make the wrong dev join the multicast
address in enable_mcast(). This causes that tipc links would
never be created.

So fix it by getting the right netdev and setting ub->ifindex,
as it does for ipv4 address.

Reported-by: Shuang Li <sh...@re...>
Signed-off-by: Xin Long <luc...@gm...>
---
 net/tipc/udp_media.c | 8 ++++++++
 1 file changed, 8 insertions(+)

diff --git a/net/tipc/udp_media.c b/net/tipc/udp_media.c
index 28a283f..9dec596 100644
--- a/net/tipc/udp_media.c
+++ b/net/tipc/udp_media.c
@@ -738,6 +738,13 @@ static int tipc_udp_enable(struct net *net, struct tipc_bearer *b,
 		b->mtu = b->media->mtu;
 #if IS_ENABLED(CONFIG_IPV6)
 	} else if (local.proto == htons(ETH_P_IPV6)) {
+		struct net_device *dev;
+
+		dev = ipv6_dev_find(net, &local.ipv6);
+		if (!dev) {
+			err = -ENODEV;
+			goto err;
+		}
 		udp_conf.family = AF_INET6;
 		udp_conf.use_udp6_tx_checksums = true;
 		udp_conf.use_udp6_rx_checksums = true;
@@ -745,6 +752,7 @@ static int tipc_udp_enable(struct net *net, struct tipc_bearer *b,
 			udp_conf.local_ip6 = in6addr_any;
 		else
 			udp_conf.local_ip6 = local.ipv6;
+		ub->ifindex = dev->ifindex;
 		b->mtu = 1280;
 #endif
 	} else {
-- 
2.1.0

[tipc-discussion] [PATCH net 1/2] ipv6: add ipv6_dev_find()

[Xin L. <luc...@gm...>]

This is to add an ip_dev_find like function for ipv6, used to find
the dev by saddr.

It will be used by TIPC protocol. So also export it.

Signed-off-by: Xin Long <luc...@gm...>
---
 include/net/addrconf.h |  2 ++
 net/ipv6/addrconf.c    | 39 +++++++++++++++++++++++++++++++++++++++
 2 files changed, 41 insertions(+)

diff --git a/include/net/addrconf.h b/include/net/addrconf.h
index 8418b7d..ba3f6c15 100644
--- a/include/net/addrconf.h
+++ b/include/net/addrconf.h
@@ -97,6 +97,8 @@ bool ipv6_chk_custom_prefix(const struct in6_addr *addr,
 
 int ipv6_chk_prefix(const struct in6_addr *addr, struct net_device *dev);
 
+struct net_device *ipv6_dev_find(struct net *net, const struct in6_addr *addr);
+
 struct inet6_ifaddr *ipv6_get_ifaddr(struct net *net,
 				     const struct in6_addr *addr,
 				     struct net_device *dev, int strict);
diff --git a/net/ipv6/addrconf.c b/net/ipv6/addrconf.c
index 840bfdb..857d6f9 100644
--- a/net/ipv6/addrconf.c
+++ b/net/ipv6/addrconf.c
@@ -1983,6 +1983,45 @@ int ipv6_chk_prefix(const struct in6_addr *addr, struct net_device *dev)
 }
 EXPORT_SYMBOL(ipv6_chk_prefix);
 
+/**
+ * ipv6_dev_find - find the first device with a given source address.
+ * @net: the net namespace
+ * @addr: the source address
+ *
+ * The caller should be protected by RCU, or RTNL.
+ */
+struct net_device *ipv6_dev_find(struct net *net, const struct in6_addr *addr)
+{
+	unsigned int hash = inet6_addr_hash(net, addr);
+	struct inet6_ifaddr *ifp, *result = NULL;
+	struct net_device *dev = NULL;
+
+	rcu_read_lock();
+	hlist_for_each_entry_rcu(ifp, &inet6_addr_lst[hash], addr_lst) {
+		if (net_eq(dev_net(ifp->idev->dev), net) &&
+		    ipv6_addr_equal(&ifp->addr, addr)) {
+			result = ifp;
+			break;
+		}
+	}
+
+	if (!result) {
+		struct rt6_info *rt;
+
+		rt = rt6_lookup(net, addr, NULL, 0, NULL, 0);
+		if (rt) {
+			dev = rt->dst.dev;
+			ip6_rt_put(rt);
+		}
+	} else {
+		dev = result->idev->dev;
+	}
+	rcu_read_unlock();
+
+	return dev;
+}
+EXPORT_SYMBOL(ipv6_dev_find);
+
 struct inet6_ifaddr *ipv6_get_ifaddr(struct net *net, const struct in6_addr *addr,
 				     struct net_device *dev, int strict)
 {
-- 
2.1.0

[tipc-discussion] [PATCH net 0/2] net: fix a mcast issue for tipc udp media

[Xin L. <luc...@gm...>]

Patch 1 is to add a function to get the dev by source address,
whcih will be used by Patch 2.

Xin Long (2):
  ipv6: add ipv6_dev_find()
  tipc: set ub->ifindex for local ipv6 address

 include/net/addrconf.h |  2 ++
 net/ipv6/addrconf.c    | 39 +++++++++++++++++++++++++++++++++++++++
 net/tipc/udp_media.c   |  8 ++++++++
 3 files changed, 49 insertions(+)

-- 
2.1.0

Re: [tipc-discussion] [PATCH net-next] tipc: Use is_broadcast_ether_addr() instead of memcmp()

[Ying X. <yin...@wi...>]

On 8/3/20 10:00 AM, Huang Guobin wrote:
> Using is_broadcast_ether_addr() instead of directly use
> memcmp() to determine if the ethernet address is broadcast
> address.
> 
> spatch with a semantic match is used to found this problem.
> (http://coccinelle.lip6.fr/)
> 
> Signed-off-by: Huang Guobin <hua...@hu...>

Acked-by: Ying Xue <yin...@wi...>

> ---
>  net/tipc/eth_media.c | 4 +---
>  1 file changed, 1 insertion(+), 3 deletions(-)
> 
> diff --git a/net/tipc/eth_media.c b/net/tipc/eth_media.c
> index 8b0bb600602d..c68019697cfe 100644
> --- a/net/tipc/eth_media.c
> +++ b/net/tipc/eth_media.c
> @@ -62,12 +62,10 @@ static int tipc_eth_raw2addr(struct tipc_bearer *b,
>  			     struct tipc_media_addr *addr,
>  			     char *msg)
>  {
> -	char bcast_mac[ETH_ALEN] = {0xff, 0xff, 0xff, 0xff, 0xff, 0xff};
> -
>  	memset(addr, 0, sizeof(*addr));
>  	ether_addr_copy(addr->value, msg);
>  	addr->media_id = TIPC_MEDIA_TYPE_ETH;
> -	addr->broadcast = !memcmp(addr->value, bcast_mac, ETH_ALEN);
> +	addr->broadcast = is_broadcast_ether_addr(addr->value);
>  	return 0;
>  }
>  
> 

[tipc-discussion] Comments on TIPC SOCK_STREAM protocol overhead?

[<lo...@ai...>]

Hello there:

Upstream TIPC's SOCK_STREAM implementation resembles SOCK_SEQPACKET; indeed, tipc_send_packet literally calls tipc_sendstream to do its job! I wonder whether the following stream protocol bandwidth issues are by design, or we could solve some or all of them:

* Lack of write coalescing usually introduces overhead in form of a small extra network frame for each send(2) unless buf_len is divisible by link MTU minus headers.
* Fragmentation for SOCK_STREAM writes caused by upper record boundary set to TIPC_MAX_USER_MSG_SIZE [1] instead of (link MTU - SHORT_H_SIZE) [2].
* pktmax - INT_H_SIZE in msg.c effectively reduces link MTU by 40 even though the accounted space is left unused.

[1] net/tipc/socket.c: send = min_t(size_t, dlen - sent, TIPC_MAX_USER_MSG_SIZE);
[2] Drawing a parallel with TCP, which does not preserve record boundaries, (link MTU - SHORT_H_SIZE) is equivalent to TCP_MAXSEG, maximum segment size.

P.S. Looks like someone saw the second issue more than a dozen years earlier, and his mail somehow slipped below the project's radar.

https://sourceforge.net/p/tipc/mailman/message/7248018/
[tipc-discussion] TIPC SOCK_STREAM performance improvement idea
From: Stephens, Allan <allan.stephens@wi...> - 2006-08-09 18:50:08

Re: [tipc-discussion] TIPC/RAFT?

[Jon M. <jm...@re...>]

(adding tipc-dek and tipc-discussion to thread)


On 7/21/20 6:04 AM, Xue, Ying wrote:
> Hi Jon,
>
> Exactly, when I learned Raft, I was always thinking how Raft and TIPC existing link/service discovery protocol can smoothly work together. Raft is a centralization consensus protocol. As you reminded below, it did match our TIPC philosophy indeed because TIPC is more close to a decentralization consensus protocol.
>
> After search, I found the famous decentralization consensus protocols include:
> https://en.wikipedia.org/wiki/Gossip_protocol
I studied the Gossip protocol before I implemented my neighbor 
monitoring ("Overlapping Ring Monitoring") algorithm. The fact is, that 
algorithm could just as well have been called a "A Biased Gossip Type 
Neighbor Monitor Algorithm", because that is what it is. In my view, 
regular Gossip is not suitable for TIPC because convergence times in 
large clusters typically become too long, and if anything changes during 
the convergence period, it will always struggle to catch up.
Also, consensus in the RAFT/PAXOS meaning is that changes in the cluster 
state must be reported with exactly the same events in exactly the same 
order on all nodes. I don't think Gossip can guarantee that any more 
than current TIPC can.
> https://en.wikipedia.org/wiki/Proof_of_work
This one seems to be used by Bitcoin. Looks interesting.
There are also a few others worth studying:
https://www.geeksforgeeks.org/consensus-algorithms-in-blockchain/


///jon

>
> It really worth learning them and I will try to research and understand which is more suitable for our case.
>
> By the way, I have to go home to take care of my kid and I could attend our today's meeting. If I have some new finding or understanding, I will keep updating.
>
> Thanks,
> Ying
>
> -----Original Message-----
> From: Jon Maloy <jm...@re...>
> Sent: Sunday, July 19, 2020 12:07 AM
> To: Xue, Ying <Yin...@wi...>
> Subject: Re: TIPC/RAFT?
>
> Hi Ying,
> I may be ranting now...
> I know that blockchain technologies (Bitcoin, Ethereum etc.) also can be used to implement consensus algorithms. Furthermore, those are completely decentralized and very scalable, something that fits much better into the TIPC philosophy than the leader electing algorithms of Paxos and Raft.
> It might be worth learning more about this before embarking on any major effort to re-implement Raft.
> I don't know much about blockchains yet, so I have no clue if this is at all feasible regarding code volumes, consensus times and other properties, but it should at least be considered before any other track is selected.
> I will try to read up on this on my side.
>
> Regards
> ///jon
>
>
> On 7/15/20 11:10 AM, Xue, Ying wrote:
>> Hi Jon,
>>
>> Raft protocol is much more complex than my initial image. The complexity mainly comes from: 1.Raft protocol itself is quite complex, so it's not very easy to totally understand it in a very detailed way; 2. Implementing a raft consensus algorithm might not be very difficult, but making sure its implementation's quality reaches an industrial criteria level is very difficult. However, for a consensus protocol, it must ensure it can work very well in any kinds of different conditions, otherwise, some distributed values become inconsistent.
>>
>> When it comes to this implementation (https://github.com/willemt/raft), I took a lot time to study, validate and understand its code so far. Although I have make all test cases contained in this implementation passed, the scenarios covered by these test cases are quite basic particularly compared to test cases included in etcd (https://github.com/etcd-io/etcd/tree/master/raft). Even in its code, it clearly comments some corner cases haven't handled yet.
>>
>> In my previous plan, I wanted to take shortcuts so that I can quickly port the implementation into Linux kernel, but it was found it's very difficult to properly adapt it into kernel space if I couldn't totally understand Raft protocol itself, which wasted my lots of time.
>>
>> Currently I have to study the following two papers:
>> https://raft.github.io/raft.pdf
>> https://github.com/ongardie/dissertation/blob/master/stanford.pdf
>>
>> While reading and understanding the papers, I am researching the code and test cases of "etcd", which can help to have a much deeper understanding of the protocol itself. Honestly, I still don't have a very well understanding for the protocol. When I read paper, I felt I had a very good understanding for some parts of the protocol, but when I came back to research its implementations, I lost again and again.
>>
>> With my understanding of the protocol becoming deeper and deeper, more and more concerns with the quality of https://github.com/willemt/raft, appear in my mind. In sum, I need more time to evaluate its quality and then consider whether we can use it as a  base to implement raft protocol in TIPC stack or if it's possible for us to well implement the protocol in kernel space within a not very long time.
>>
>> Although I have to admit studying and even implementing the protocol is a pretty challenging thing, I am still quite interested in continuing working on this topic.
>>
>> Thanks,
>> Ying
>>
>> -----Original Message-----
>> From: Jon Maloy <jm...@re...>
>> Sent: Saturday, July 11, 2020 2:29 AM
>> To: Xue, Ying <Yin...@wi...>
>> Subject: TIPC/RAFT?
>>
>> Hi Ying,
>> Since I haven't had any news from you for a while, I was just getting curious.
>> Did you have any time to look into the RAFT implementation references I sent you?
>> Do you have any thougths about this?
>>
>> BR
>> ///jon
>>

Re: [tipc-discussion] [PATCH RFC 3/5] tipc: introduce encryption master key

[Jon M. <jm...@re...>]

On 7/21/20 6:33 AM, Tuong Tong Lien wrote:
> Hi Jon,
>
> Thanks a lot for your comments, I will update them accordingly!
> Regarding the 'master key' definition, I have just done a search and found it to perfectly matched our case:
>
> --------------
> https://en.wikipedia.org/wiki/Cryptographic_key_types
> Symmetric master key
>      A symmetric master key is used to derive other symmetric keys (e.g., data encryption keys, key wrapping keys, or authentication keys) using symmetric cryptographic methods.
> --------------
>
> I think it is fine to go with it, of course unless we find a better one... 😊!
>
> BR/Tuong
Pefect! Let's go for that. Maybe no need to update the code, just the 
logs and descriptions.

///jon

>
> -----Original Message-----
> From: Jon Maloy <jm...@re...>
> Sent: Tuesday, July 21, 2020 1:07 AM
> To: Tuong Tong Lien <tuo...@de...>; ma...@do...; yin...@wi...; tip...@li...
> Cc: tipc-dek <tip...@de...>
> Subject: Re: [PATCH RFC 3/5] tipc: introduce encryption master key
>
>
>
> On 7/10/20 6:11 AM, Tuong Lien wrote:
>> In addition to the supported cluster & per-node encryption keys for the
>> en/decryption of TIPC messages, we now introduce one option for user to
>> set a cluster key as 'master key',
> Still not quite happy with the term 'master key' but have no really good
> suggestions at the moment.
> Maybe 'membership key' or just 'member key'?
>> which is simply a symmetric key like
>> the former
> but has a longer life cycle. It has two purposes:
>
>    - Authentication of new member nodes in the cluster.
>     New nodes, having no knowledge of current session keys
>     in the cluster will still be able to join the cluster as long as
>     they know this key. This is because all neighbor discovery
>     (LINK_CONFIG) messages must be encrypted with this key.
>
> - Encryption of session encryption keys during automatic exchange
>     and update of those. This is  a feature we will introduce
>     in a later commit in this series.
> We insert the new key into the currently unused slot 0 in the key array,
> and start using it immediately once the user has set it.
> After joining, a node only knowing the master key should be fully
>> communicable to existing nodes in the cluster, although those nodes may
>> have their own session keys activated (i.e. not the master one). To
>> support this, we simply
> s/simply//
>> define a 'grace period', starting from the time
>> a node itself reports having no RX keys, so the existing nodes will use
>> the master key for encryption instead. The grace period can be extended
>> but will automatically stop after, e.g. 5 seconds without a new report.
>> This is also the basis for later key exchanging since the new node will
>> be impossible to decrypt anything without the support from master key.
>>
>> For user to set a master key, we define a new netlink flag -
>> 'TIPC_NLA_NODE_KEY_MASTER', so it can be added to the current 'set key'
>> netlink command to specify the setting key to be a master key.
>>
>> Above all, the traditional cluster/per-node key mechanism is guaranteed
>> to work when user comes not to use this master key option. This is also
>> compatible to legacy nodes without the feature supported.
> Even this new key can be updated without any interruption
> of cluster connectivity, but is so is needed, this has to be
> coordinated and set by the user.
>> Signed-off-by: Tuong Lien <tuo...@de...>
>> ---
>>    include/uapi/linux/tipc_netlink.h |   1 +
>>    net/tipc/crypto.c                 | 206 ++++++++++++++++++++++++++++----------
>>    net/tipc/crypto.h                 |  15 ++-
>>    net/tipc/msg.h                    |   4 +-
>>    net/tipc/netlink.c                |   1 +
>>    net/tipc/node.c                   |  46 ++++-----
>>    6 files changed, 189 insertions(+), 84 deletions(-)
>>
>> diff --git a/include/uapi/linux/tipc_netlink.h b/include/uapi/linux/tipc_netlink.h
>> index dc0d23a50e69..d484baa9d365 100644
>> --- a/include/uapi/linux/tipc_netlink.h
>> +++ b/include/uapi/linux/tipc_netlink.h
>> @@ -165,6 +165,7 @@ enum {
>>    	TIPC_NLA_NODE_UP,		/* flag */
>>    	TIPC_NLA_NODE_ID,		/* data */
>>    	TIPC_NLA_NODE_KEY,		/* data */
>> +	TIPC_NLA_NODE_KEY_MASTER,	/* flag */
>>    
>>    	__TIPC_NLA_NODE_MAX,
>>    	TIPC_NLA_NODE_MAX = __TIPC_NLA_NODE_MAX - 1
>> diff --git a/net/tipc/crypto.c b/net/tipc/crypto.c
>> index f1046ab4bd01..7c688cd0b13e 100644
>> --- a/net/tipc/crypto.c
>> +++ b/net/tipc/crypto.c
>> @@ -38,6 +38,7 @@
>>    #include <crypto/aes.h>
>>    #include "crypto.h"
>>    
>> +#define TIPC_TX_GRACE_PERIOD	msecs_to_jiffies(5000) /* 5s */
>>    #define TIPC_TX_LASTING_TIME	msecs_to_jiffies(10000) /* 10s */
>>    #define TIPC_RX_ACTIVE_LIM	msecs_to_jiffies(3000) /* 3s */
>>    #define TIPC_RX_PASSIVE_LIM	msecs_to_jiffies(15000) /* 15s */
>> @@ -49,9 +50,9 @@
>>     * TIPC Key ids
>>     */
>>    enum {
>> -	KEY_UNUSED = 0,
>> -	KEY_MIN,
>> -	KEY_1 = KEY_MIN,
>> +	KEY_MASTER = 0,
>> +	KEY_MIN = KEY_MASTER,
>> +	KEY_1 = 1,
>>    	KEY_2,
>>    	KEY_3,
>>    	KEY_MAX = KEY_3,
>> @@ -166,27 +167,36 @@ struct tipc_crypto_stats {
>>     * @aead: array of pointers to AEAD keys for encryption/decryption
>>     * @peer_rx_active: replicated peer RX active key index
>>     * @key: the key states
>> - * @working: the crypto is working or not
>>     * @stats: the crypto statistics
>>     * @name: the crypto name
>>     * @sndnxt: the per-peer sndnxt (TX)
>>     * @timer1: general timer 1 (jiffies)
>>     * @timer2: general timer 2 (jiffies)
>> + * @working: the crypto is working or not
>> + * @key_master: flag indicates if master key exists
>> + * @legacy_user: flag indicates if a peer joins w/o master key (for bwd comp.)
>>     * @lock: tipc_key lock
>>     */
>>    struct tipc_crypto {
>>    	struct net *net;
>>    	struct tipc_node *node;
>> -	struct tipc_aead __rcu *aead[KEY_MAX + 1]; /* key[0] is UNUSED */
>> +	struct tipc_aead __rcu *aead[KEY_MAX + 1];
>>    	atomic_t peer_rx_active;
>>    	struct tipc_key key;
>> -	u8 working:1;
>>    	struct tipc_crypto_stats __percpu *stats;
>>    	char name[48];
>>    
>>    	atomic64_t sndnxt ____cacheline_aligned;
>>    	unsigned long timer1;
>>    	unsigned long timer2;
>> +	union {
>> +		struct {
>> +			u8 working:1;
>> +			u8 key_master:1;
>> +			u8 legacy_user:1;
>> +		};
>> +		u8 flags;
>> +	};
>>    	spinlock_t lock; /* crypto lock */
>>    
>>    } ____cacheline_aligned;
>> @@ -236,13 +246,19 @@ static inline void tipc_crypto_key_set_state(struct tipc_crypto *c,
>>    					     u8 new_active,
>>    					     u8 new_pending);
>>    static int tipc_crypto_key_attach(struct tipc_crypto *c,
>> -				  struct tipc_aead *aead, u8 pos);
>> +				  struct tipc_aead *aead, u8 pos,
>> +				  bool master_key);
>>    static bool tipc_crypto_key_try_align(struct tipc_crypto *rx, u8 new_pending);
>>    static struct tipc_aead *tipc_crypto_key_pick_tx(struct tipc_crypto *tx,
>>    						 struct tipc_crypto *rx,
>> -						 struct sk_buff *skb);
>> +						 struct sk_buff *skb,
>> +						 u8 tx_key);
>>    static void tipc_crypto_key_synch(struct tipc_crypto *rx, struct sk_buff *skb);
>>    static int tipc_crypto_key_revoke(struct net *net, u8 tx_key);
>> +static inline void tipc_crypto_clone_msg(struct net *net, struct sk_buff *_skb,
>> +					 struct tipc_bearer *b,
>> +					 struct tipc_media_addr *dst,
>> +					 struct tipc_node *__dnode, u8 type);
>>    static void tipc_crypto_rcv_complete(struct net *net, struct tipc_aead *aead,
>>    				     struct tipc_bearer *b,
>>    				     struct sk_buff **skb, int err);
>> @@ -937,8 +953,6 @@ bool tipc_ehdr_validate(struct sk_buff *skb)
>>    		return false;
>>    	if (unlikely(skb->len <= ehsz + TIPC_AES_GCM_TAG_SIZE))
>>    		return false;
>> -	if (unlikely(!ehdr->tx_key))
>> -		return false;
>>    
>>    	return true;
>>    }
>> @@ -991,6 +1005,8 @@ static int tipc_ehdr_build(struct net *net, struct tipc_aead *aead,
>>    	ehdr->tx_key = tx_key;
>>    	ehdr->destined = (__rx) ? 1 : 0;
>>    	ehdr->rx_key_active = (__rx) ? __rx->key.active : 0;
>> +	ehdr->rx_nokey = (__rx) ? !__rx->key.keys : 0;
>> +	ehdr->master_key = aead->crypto->key_master;
>>    	ehdr->reserved_1 = 0;
>>    	ehdr->reserved_2 = 0;
>>    
>> @@ -1033,6 +1049,7 @@ static inline void tipc_crypto_key_set_state(struct tipc_crypto *c,
>>     * @c: TIPC crypto to which new key is attached
>>     * @ukey: the user key
>>     * @mode: the key mode (CLUSTER_KEY or PER_NODE_KEY)
>> + * @master_key: specify this is a cluster master key
>>     *
>>     * A new TIPC AEAD key will be allocated and initiated with the specified user
>>     * key, then attached to the TIPC crypto.
>> @@ -1040,7 +1057,7 @@ static inline void tipc_crypto_key_set_state(struct tipc_crypto *c,
>>     * Return: new key id in case of success, otherwise: < 0
>>     */
>>    int tipc_crypto_key_init(struct tipc_crypto *c, struct tipc_aead_key *ukey,
>> -			 u8 mode)
>> +			 u8 mode, bool master_key)
>>    {
>>    	struct tipc_aead *aead = NULL;
>>    	int rc = 0;
>> @@ -1053,7 +1070,7 @@ int tipc_crypto_key_init(struct tipc_crypto *c, struct tipc_aead_key *ukey,
>>    	}
>>    
>>    	/* Attach it to the crypto */
>> -	rc = tipc_crypto_key_attach(c, aead, 0);
>> +	rc = tipc_crypto_key_attach(c, aead, 0, master_key);
>>    	if (rc < 0) {
>>    		pr_err("%s: unable to attach key, err %d\n", c->name, rc);
>>    		tipc_aead_free(&aead->rcu);
>> @@ -1069,11 +1086,13 @@ int tipc_crypto_key_init(struct tipc_crypto *c, struct tipc_aead_key *ukey,
>>     * @c: TIPC crypto to which the new AEAD key is attached
>>     * @aead: the new AEAD key pointer
>>     * @pos: desired slot in the crypto key array, = 0 if any!
>> + * @master_key: specify this is a cluster master key
>>     *
>>     * Return: new key id in case of success, otherwise: -EBUSY
>>     */
>>    static int tipc_crypto_key_attach(struct tipc_crypto *c,
>> -				  struct tipc_aead *aead, u8 pos)
>> +				  struct tipc_aead *aead, u8 pos,
>> +				  bool master_key)
>>    {
>>    	struct tipc_key key;
>>    	int rc = -EBUSY;
>> @@ -1081,6 +1100,10 @@ static int tipc_crypto_key_attach(struct tipc_crypto *c,
>>    
>>    	spin_lock_bh(&c->lock);
>>    	key = c->key;
>> +	if (master_key) {
>> +		new_key = KEY_MASTER;
>> +		goto attach;
>> +	}
>>    	if (key.active && key.passive)
>>    		goto exit;
>>    	if (key.pending) {
>> @@ -1112,8 +1135,7 @@ static int tipc_crypto_key_attach(struct tipc_crypto *c,
>>    		tipc_crypto_key_set_state(c, key.passive, key.active,
>>    					  key.pending);
>>    	c->working = 1;
>> -	c->timer1 = jiffies;
>> -	c->timer2 = jiffies;
>> +	c->key_master |= master_key;
>>    	rc = new_key;
>>    
>>    exit:
>> @@ -1126,7 +1148,7 @@ void tipc_crypto_key_flush(struct tipc_crypto *c)
>>    	int k;
>>    
>>    	spin_lock_bh(&c->lock);
>> -	c->working = 0;
>> +	c->flags = 0;
>>    	tipc_crypto_key_set_state(c, 0, 0, 0);
>>    	for (k = KEY_MIN; k <= KEY_MAX; k++)
>>    		tipc_crypto_key_detach(c->aead[k], &c->lock);
>> @@ -1202,6 +1224,7 @@ static bool tipc_crypto_key_try_align(struct tipc_crypto *rx, u8 new_pending)
>>     * @tx: TX crypto handle
>>     * @rx: RX crypto handle (can be NULL)
>>     * @skb: the message skb which will be decrypted later
>> + * @tx_key: peer TX key id
>>     *
>>     * This function looks up the existing TX keys and pick one which is suitable
>>     * for the message decryption, that must be a cluster key and not used before
>> @@ -1211,7 +1234,8 @@ static bool tipc_crypto_key_try_align(struct tipc_crypto *rx, u8 new_pending)
>>     */
>>    static struct tipc_aead *tipc_crypto_key_pick_tx(struct tipc_crypto *tx,
>>    						 struct tipc_crypto *rx,
>> -						 struct sk_buff *skb)
>> +						 struct sk_buff *skb,
>> +						 u8 tx_key)
>>    {
>>    	struct tipc_skb_cb *skb_cb = TIPC_SKB_CB(skb);
>>    	struct tipc_aead *aead = NULL;
>> @@ -1230,6 +1254,10 @@ static struct tipc_aead *tipc_crypto_key_pick_tx(struct tipc_crypto *tx,
>>    
>>    	/* Pick one TX key */
>>    	spin_lock(&tx->lock);
>> +	if (tx_key == KEY_MASTER) {
>> +		aead = tipc_aead_rcu_ptr(tx->aead[KEY_MASTER], &tx->lock);
>> +		goto done;
>> +	}
>>    	do {
>>    		k = (i == 0) ? key.pending :
>>    			((i == 1) ? key.active : key.passive);
>> @@ -1249,9 +1277,12 @@ static struct tipc_aead *tipc_crypto_key_pick_tx(struct tipc_crypto *tx,
>>    		skb->next = skb_clone(skb, GFP_ATOMIC);
>>    		if (unlikely(!skb->next))
>>    			pr_warn("Failed to clone skb for next round if any\n");
>> -		WARN_ON(!refcount_inc_not_zero(&aead->refcnt));
>>    		break;
>>    	} while (++i < 3);
>> +
>> +done:
>> +	if (likely(aead))
>> +		WARN_ON(!refcount_inc_not_zero(&aead->refcnt));
>>    	spin_unlock(&tx->lock);
>>    
>>    	return aead;
>> @@ -1266,6 +1297,9 @@ static struct tipc_aead *tipc_crypto_key_pick_tx(struct tipc_crypto *tx,
>>     * has changed, so the number of TX keys' users on this node are increased and
>>     * decreased correspondingly.
>>     *
>> + * It also considers if peer has no key, then we need to make own master key
>> + * (if any) taking over i.e. starting grace period.
>> + *
>>     * The "per-peer" sndnxt is also reset when the peer key has switched.
>>     */
>>    static void tipc_crypto_key_synch(struct tipc_crypto *rx, struct sk_buff *skb)
>> @@ -1276,11 +1310,23 @@ static void tipc_crypto_key_synch(struct tipc_crypto *rx, struct sk_buff *skb)
>>    	u32 self = tipc_own_addr(rx->net);
>>    	u8 cur, new;
>>    
>> -	/* Ensure this message is destined to us first */
>> +	/* Update RX 'key_master' flag according to peer, also mark "legacy" if
>> +	 * a peer has no master key.
>> +	 */
>> +	rx->key_master = ehdr->master_key;
>> +	if (!rx->key_master)
>> +		tx->legacy_user = 1;
>> +
>> +	/* For later cases, apply only if message is destined to this node */
>>    	if (!ehdr->destined || msg_short(hdr) || msg_destnode(hdr) != self)
>>    		return;
>>    
>> -	/* Peer RX active key has changed, let's update own TX users */
>> +	/* Case 1: Peer has no keys, let's make master key taking over */
> s/taking/take/
>> +	if (ehdr->rx_nokey)
>> +		/* Set or extend grace period */
>> +		tx->timer2 = jiffies;
>> +
>> +	/* Case 2: Peer RX active key has changed, let's update own TX users */
>>    	cur = atomic_read(&rx->peer_rx_active);
>>    	new = ehdr->rx_key_active;
>>    	if (tx->key.keys &&
>> @@ -1338,7 +1384,7 @@ int tipc_crypto_start(struct tipc_crypto **crypto, struct net *net,
>>    		return -ENOMEM;
>>    	}
>>    
>> -	c->working = 0;
>> +	c->flags = 0;
>>    	c->net = net;
>>    	c->node = node;
>>    	tipc_crypto_key_set_state(c, 0, 0, 0);
>> @@ -1473,6 +1519,12 @@ void tipc_crypto_timeout(struct tipc_crypto *rx)
>>    s5:
>>    	spin_unlock(&rx->lock);
>>    
>> +	/* Relax it here, the flag will be set again if it really is, but only
>> +	 * when we are not in grace period for safety!
>> +	 */
>> +	if (time_after(jiffies, tx->timer2 + TIPC_TX_GRACE_PERIOD))
>> +		tx->legacy_user = 0;
>> +
>>    	/* Limit max_tfms & do debug commands if needed */
>>    	if (likely(sysctl_tipc_max_tfms <= TIPC_MAX_TFMS_LIM))
>>    		return;
>> @@ -1482,6 +1534,22 @@ void tipc_crypto_timeout(struct tipc_crypto *rx)
>>    	tipc_crypto_do_cmd(rx->net, cmd);
>>    }
>>    
>> +static inline void tipc_crypto_clone_msg(struct net *net, struct sk_buff *_skb,
>> +					 struct tipc_bearer *b,
>> +					 struct tipc_media_addr *dst,
>> +					 struct tipc_node *__dnode, u8 type)
>> +{
>> +	struct sk_buff *skb;
>> +
>> +	skb = skb_clone(_skb, GFP_ATOMIC);
>> +	if (skb) {
>> +		TIPC_SKB_CB(skb)->xmit_type = type;
>> +		tipc_crypto_xmit(net, &skb, b, dst, __dnode);
>> +		if (skb)
>> +			b->media->send_msg(net, skb, b, dst);
>> +	}
>> +}
>> +
>>    /**
>>     * tipc_crypto_xmit - Build & encrypt TIPC message for xmit
>>     * @net: struct net
>> @@ -1491,7 +1559,8 @@ void tipc_crypto_timeout(struct tipc_crypto *rx)
>>     * @__dnode: destination node for reference if any
>>     *
>>     * First, build an encryption message header on the top of the message, then
>> - * encrypt the original TIPC message by using the active or pending TX key.
>> + * encrypt the original TIPC message by using the pending, master or active
>> + * key with this preference order.
>>     * If the encryption is successful, the encrypted skb is returned directly or
>>     * via the callback.
>>     * Otherwise, the skb is freed!
>> @@ -1514,46 +1583,63 @@ int tipc_crypto_xmit(struct net *net, struct sk_buff **skb,
>>    	struct tipc_msg *hdr = buf_msg(*skb);
>>    	struct tipc_key key = tx->key;
>>    	struct tipc_aead *aead = NULL;
>> -	struct sk_buff *_skb;
>> -	int rc = -ENOKEY;
>>    	u32 user = msg_user(hdr);
>> -	u8 tx_key;
>> +	u32 type = msg_type(hdr);
>> +	int rc = -ENOKEY;
>> +	u8 tx_key = 0;
>>    
>>    	/* No encryption? */
>>    	if (!tx->working)
>>    		return 0;
>>    
>> -	/* Try with the pending key if available and:
>> -	 * 1) This is the only choice (i.e. no active key) or;
>> -	 * 2) Peer has switched to this key (unicast only) or;
>> -	 * 3) It is time to do a pending key probe;
>> -	 */
>> +	/* Pending key if peer has active on it or probing time */
>>    	if (unlikely(key.pending)) {
>>    		tx_key = key.pending;
>> -		if (!key.active)
>> +		if (!tx->key_master && !key.active)
>>    			goto encrypt;
>>    		if (__rx && atomic_read(&__rx->peer_rx_active) == tx_key)
>>    			goto encrypt;
>> -		if (TIPC_SKB_CB(*skb)->probe) {
>> +		if (TIPC_SKB_CB(*skb)->xmit_type == SKB_PROBING) {
>>    			pr_debug("%s: probing for key[%d]\n", tx->name,
>>    				 key.pending);
>>    			goto encrypt;
>>    		}
>> -		if (user == LINK_CONFIG || user == LINK_PROTOCOL) {
>> -			_skb = skb_clone(*skb, GFP_ATOMIC);
>> -			if (_skb) {
>> -				TIPC_SKB_CB(_skb)->probe = 1;
>> -				tipc_crypto_xmit(net, &_skb, b, dst, __dnode);
>> -				if (_skb)
>> -					b->media->send_msg(net, _skb, b, dst);
>> +		if (user == LINK_CONFIG || user == LINK_PROTOCOL)
>> +			tipc_crypto_clone_msg(net, *skb, b, dst, __dnode,
>> +					      SKB_PROBING);
>> +	}
>> +
>> +	/* Master key if this is a *vital* message or in grace period */
>> +	if (tx->key_master) {
>> +		tx_key = KEY_MASTER;
>> +		if (!key.active)
>> +			goto encrypt;
>> +		if (TIPC_SKB_CB(*skb)->xmit_type == SKB_GRACING) {
>> +			pr_debug("%s: gracing for msg (%d %d)\n", tx->name,
>> +				 user, type);
>> +			goto encrypt;
>> +		}
>> +		if (user == LINK_CONFIG ||
>> +		    (user == LINK_PROTOCOL && type == RESET_MSG) ||
>> +		    time_before(jiffies, tx->timer2 + TIPC_TX_GRACE_PERIOD)) {
> Why do RESET messages need to be encrypted with the master key?
>> +			if (__rx && __rx->key_master &&
>> +			    !atomic_read(&__rx->peer_rx_active))
>> +				goto encrypt;
>> +			if (!__rx) {
>> +				if (likely(!tx->legacy_user))
>> +					goto encrypt;
>> +				tipc_crypto_clone_msg(net, *skb, b, dst,
>> +						      __dnode, SKB_GRACING);
>>    			}
>>    		}
>>    	}
>> +
>>    	/* Else, use the active key if any */
>>    	if (likely(key.active)) {
>>    		tx_key = key.active;
>>    		goto encrypt;
>>    	}
>> +
>>    	goto exit;
>>    
>>    encrypt:
>> @@ -1619,15 +1705,16 @@ int tipc_crypto_rcv(struct net *net, struct tipc_crypto *rx,
>>    	struct tipc_aead *aead = NULL;
>>    	struct tipc_key key;
>>    	int rc = -ENOKEY;
>> -	u8 tx_key = 0;
>> +	u8 tx_key;
>> +
>> +	tx_key = ((struct tipc_ehdr *)(*skb)->data)->tx_key;
>>    
>>    	/* New peer?
>>    	 * Let's try with TX key (i.e. cluster mode) & verify the skb first!
>>    	 */
>> -	if (unlikely(!rx))
>> +	if (unlikely(!rx || tx_key == KEY_MASTER))
>>    		goto pick_tx;
>>    
>> -	tx_key = ((struct tipc_ehdr *)(*skb)->data)->tx_key;
>>    	/* Pick RX key according to TX key if any */
>>    	key = rx->key;
>>    	if (tx_key == key.active || tx_key == key.pending ||
>> @@ -1640,7 +1727,7 @@ int tipc_crypto_rcv(struct net *net, struct tipc_crypto *rx,
>>    
>>    pick_tx:
>>    	/* No key suitable? Try to pick one from TX... */
>> -	aead = tipc_crypto_key_pick_tx(tx, rx, *skb);
>> +	aead = tipc_crypto_key_pick_tx(tx, rx, *skb, tx_key);
>>    	if (aead)
>>    		goto decrypt;
>>    	goto exit;
>> @@ -1722,9 +1809,12 @@ static void tipc_crypto_rcv_complete(struct net *net, struct tipc_aead *aead,
>>    				goto free_skb;
>>    		}
>>    
>> +		/* Ignore cloning if it was TX master key */
>> +		if (ehdr->tx_key == KEY_MASTER)
>> +			goto rcv;
>>    		if (tipc_aead_clone(&tmp, aead) < 0)
>>    			goto rcv;
>> -		if (tipc_crypto_key_attach(rx, tmp, ehdr->tx_key) < 0) {
>> +		if (tipc_crypto_key_attach(rx, tmp, ehdr->tx_key, false) < 0) {
>>    			tipc_aead_free(&tmp->rcu);
>>    			goto rcv;
>>    		}
>> @@ -1740,10 +1830,10 @@ static void tipc_crypto_rcv_complete(struct net *net, struct tipc_aead *aead,
>>    	/* Set the RX key's user */
>>    	tipc_aead_users_set(aead, 1);
>>    
>> -rcv:
>>    	/* Mark this point, RX works */
>>    	rx->timer1 = jiffies;
>>    
>> +rcv:
>>    	/* Remove ehdr & auth. tag prior to tipc_rcv() */
>>    	ehdr = (struct tipc_ehdr *)(*skb)->data;
>>    
>> @@ -1865,14 +1955,24 @@ static char *tipc_crypto_key_dump(struct tipc_crypto *c, char *buf)
>>    	char *s;
>>    
>>    	for (k = KEY_MIN; k <= KEY_MAX; k++) {
>> -		if (k == key.passive)
>> -			s = "PAS";
>> -		else if (k == key.active)
>> -			s = "ACT";
>> -		else if (k == key.pending)
>> -			s = "PEN";
>> -		else
>> -			s = "-";
>> +		if (k == KEY_MASTER) {
>> +			if (is_rx(c))
>> +				continue;
>> +			if (time_before(jiffies,
>> +					c->timer2 + TIPC_TX_GRACE_PERIOD))
>> +				s = "ACT";
>> +			else
>> +				s = "PAS";
>> +		} else {
>> +			if (k == key.passive)
>> +				s = "PAS";
>> +			else if (k == key.active)
>> +				s = "ACT";
>> +			else if (k == key.pending)
>> +				s = "PEN";
>> +			else
>> +				s = "-";
>> +		}
>>    		i += scnprintf(buf + i, 200 - i, "\tKey%d: %s", k, s);
>>    
>>    		rcu_read_lock();
>> diff --git a/net/tipc/crypto.h b/net/tipc/crypto.h
>> index c3de769f49e8..7fcb80cb0e8a 100644
>> --- a/net/tipc/crypto.h
>> +++ b/net/tipc/crypto.h
>> @@ -74,7 +74,7 @@ extern int sysctl_tipc_max_tfms __read_mostly;
>>     *     3 3 2 2 2 2 2 2 2 2 2 2 1 1 1 1 1 1 1 1 1 1 0 0 0 0 0 0 0 0 0 0
>>     *     1 0 9 8 7 6 5 4|3 2 1 0 9 8 7 6|5 4 3 2 1 0 9 8|7 6 5 4 3 2 1 0
>>     *    +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
>> - * w0:|Ver=7| User  |D|TX |RX |K|                 Rsvd                |
>> + * w0:|Ver=7| User  |D|TX |RX |K|M|N|             Rsvd                |
>>     *    +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
>>     * w1:|                             Seqno                             |
>>     * w2:|                           (8 octets)                          |
>> @@ -101,6 +101,9 @@ extern int sysctl_tipc_max_tfms __read_mostly;
>>     *	RX	: Currently RX active key corresponding to the destination
>>     *	          node's TX key (when the "D" bit is set)
>>     *	K	: Keep-alive bit (for RPS, LINK_PROTOCOL/STATE_MSG only)
>> + *	M       : Bit indicates if sender has master key
>> + *	N	: Bit indicates if sender has no RX keys corresponding to the
>> + *	          receiver's TX (when the "D" bit is set)
>>     *	Rsvd	: Reserved bit, field
>>     * Word1-2:
>>     *	Seqno	: The 64-bit sequence number of the encrypted message, also
>> @@ -117,7 +120,9 @@ struct tipc_ehdr {
>>    			__u8	destined:1,
>>    				user:4,
>>    				version:3;
>> -			__u8	reserved_1:3,
>> +			__u8	reserved_1:1,
>> +				rx_nokey:1,
>> +				master_key:1,
>>    				keepalive:1,
>>    				rx_key_active:2,
>>    				tx_key:2;
>> @@ -128,7 +133,9 @@ struct tipc_ehdr {
>>    			__u8	tx_key:2,
>>    				rx_key_active:2,
>>    				keepalive:1,
>> -				reserved_1:3;
>> +				master_key:1,
>> +				rx_nokey:1,
>> +				reserved_1:1;
>>    #else
>>    #error  "Please fix <asm/byteorder.h>"
>>    #endif
>> @@ -158,7 +165,7 @@ int tipc_crypto_xmit(struct net *net, struct sk_buff **skb,
>>    int tipc_crypto_rcv(struct net *net, struct tipc_crypto *rx,
>>    		    struct sk_buff **skb, struct tipc_bearer *b);
>>    int tipc_crypto_key_init(struct tipc_crypto *c, struct tipc_aead_key *ukey,
>> -			 u8 mode);
>> +			 u8 mode, bool master_key);
>>    void tipc_crypto_key_flush(struct tipc_crypto *c);
>>    int tipc_aead_key_validate(struct tipc_aead_key *ukey);
>>    bool tipc_ehdr_validate(struct sk_buff *skb);
>> diff --git a/net/tipc/msg.h b/net/tipc/msg.h
>> index 1016e96db5c4..25e5c5c8a6ff 100644
>> --- a/net/tipc/msg.h
>> +++ b/net/tipc/msg.h
>> @@ -127,7 +127,9 @@ struct tipc_skb_cb {
>>    #ifdef CONFIG_TIPC_CRYPTO
>>    			u8 encrypted:1;
>>    			u8 decrypted:1;
>> -			u8 probe:1;
>> +#define SKB_PROBING	1
>> +#define SKB_GRACING	2
>> +			u8 xmit_type:2;
>>    			u8 tx_clone_deferred:1;
>>    #endif
>>    		};
>> diff --git a/net/tipc/netlink.c b/net/tipc/netlink.c
>> index c4aee6247d55..1ec00fcc26ee 100644
>> --- a/net/tipc/netlink.c
>> +++ b/net/tipc/netlink.c
>> @@ -108,6 +108,7 @@ const struct nla_policy tipc_nl_node_policy[TIPC_NLA_NODE_MAX + 1] = {
>>    					    .len = TIPC_NODEID_LEN},
>>    	[TIPC_NLA_NODE_KEY]		= { .type = NLA_BINARY,
>>    					    .len = TIPC_AEAD_KEY_SIZE_MAX},
>> +	[TIPC_NLA_NODE_KEY_MASTER]	= { .type = NLA_FLAG },
>>    };
>>    
>>    /* Properties valid for media, bearer and link */
>> diff --git a/net/tipc/node.c b/net/tipc/node.c
>> index 030a51c4d1fa..55f012d1ea74 100644
>> --- a/net/tipc/node.c
>> +++ b/net/tipc/node.c
>> @@ -2872,11 +2872,11 @@ static int __tipc_nl_node_set_key(struct sk_buff *skb, struct genl_info *info)
>>    {
>>    	struct nlattr *attrs[TIPC_NLA_NODE_MAX + 1];
>>    	struct net *net = sock_net(skb->sk);
>> -	struct tipc_net *tn = tipc_net(net);
>> +	struct tipc_crypto *tx = tipc_net(net)->crypto_tx, *c = tx;
>>    	struct tipc_node *n = NULL;
>>    	struct tipc_aead_key *ukey;
>> -	struct tipc_crypto *c;
>> -	u8 *id, *own_id;
>> +	bool master_key = false;
>> +	u8 *id, *own_id, mode;
>>    	int rc = 0;
>>    
>>    	if (!info->attrs[TIPC_NLA_NODE])
>> @@ -2886,51 +2886,45 @@ static int __tipc_nl_node_set_key(struct sk_buff *skb, struct genl_info *info)
>>    			      info->attrs[TIPC_NLA_NODE],
>>    			      tipc_nl_node_policy, info->extack);
>>    	if (rc)
>> -		goto exit;
>> +		return rc;
>>    
>>    	own_id = tipc_own_id(net);
>> -	if (!own_id) {
>> -		rc = -EPERM;
>> -		goto exit;
>> -	}
>> +	if (!own_id)
>> +		return -EPERM;
>>    
>>    	rc = tipc_nl_retrieve_key(attrs, &ukey);
>>    	if (rc)
>> -		goto exit;
>> +		return rc;
>>    
>>    	rc = tipc_aead_key_validate(ukey);
>>    	if (rc)
>> -		goto exit;
>> +		return rc;
>>    
>>    	rc = tipc_nl_retrieve_nodeid(attrs, &id);
>>    	switch (rc) {
>>    	case -ENODATA:
>> -		/* Cluster key mode */
>> -		rc = tipc_crypto_key_init(tn->crypto_tx, ukey, CLUSTER_KEY);
>> +		mode = CLUSTER_KEY;
>> +		master_key = !!(attrs[TIPC_NLA_NODE_KEY_MASTER]);
>>    		break;
>>    	case 0:
>> -		/* Per-node key mode */
>> -		if (!memcmp(id, own_id, NODE_ID_LEN)) {
>> -			c = tn->crypto_tx;
>> -		} else {
>> +		mode = PER_NODE_KEY;
>> +		if (memcmp(id, own_id, NODE_ID_LEN)) {
>>    			n = tipc_node_find_by_id(net, id) ?:
>>    				tipc_node_create(net, 0, id, 0xffffu, 0, true);
>> -			if (unlikely(!n)) {
>> -				rc = -ENOMEM;
>> -				break;
>> -			}
>> +			if (unlikely(!n))
>> +				return -ENOMEM;
>>    			c = n->crypto_rx;
>>    		}
>> -
>> -		rc = tipc_crypto_key_init(c, ukey, PER_NODE_KEY);
>> -		if (n)
>> -			tipc_node_put(n);
>>    		break;
>>    	default:
>> -		break;
>> +		return rc;
>>    	}
>>    
>> -exit:
>> +	/* Initiate the TX/RX key */
>> +	rc = tipc_crypto_key_init(c, ukey, mode, master_key);
>> +	if (n)
>> +		tipc_node_put(n);
>> +
>>    	return (rc < 0) ? rc : 0;
>>    }
>>    
> Acked-by: Jon Maloy <jm...@re...>
>

Re: [tipc-discussion] [net-next] tipc: fix NULL pointer dereference in streaming

[Xin L. <luc...@gm...>]

On Tue, Jul 21, 2020 at 7:26 PM Tuong Tong Lien
<tuo...@de...> wrote:
>
>
>
> > -----Original Message-----
> > From: Xin Long <luc...@gm...>
> > Sent: Tuesday, July 21, 2020 6:23 PM
> > To: Tuong Tong Lien <tuo...@de...>
> > Cc: davem <da...@da...>; jm...@re...; ma...@do...; Ying Xue <yin...@wi...>; network dev
> > <ne...@vg...>; tip...@li...
> > Subject: Re: [tipc-discussion] [net-next] tipc: fix NULL pointer dereference in streaming
> >
> > On Wed, Jun 3, 2020 at 1:06 PM Tuong Lien <tuo...@de...> wrote:
> > >
> > > syzbot found the following crash:
> > >
> > > general protection fault, probably for non-canonical address 0xdffffc0000000019: 0000 [#1] PREEMPT SMP KASAN
> > > KASAN: null-ptr-deref in range [0x00000000000000c8-0x00000000000000cf]
> > > CPU: 1 PID: 7060 Comm: syz-executor394 Not tainted 5.7.0-rc6-syzkaller #0
> > > Hardware name: Google Google Compute Engine/Google Compute Engine, BIOS Google 01/01/2011
> > > RIP: 0010:__tipc_sendstream+0xbde/0x11f0 net/tipc/socket.c:1591
> > > Code: 00 00 00 00 48 39 5c 24 28 48 0f 44 d8 e8 fa 3e db f9 48 b8 00 00 00 00 00 fc ff df 48 8d bb c8 00 00 00 48 89 fa 48 c1 ea 03 <80> 3c
> > 02 00 0f 85 e2 04 00 00 48 8b 9b c8 00 00 00 48 b8 00 00 00
> > > RSP: 0018:ffffc90003ef7818 EFLAGS: 00010202
> > > RAX: dffffc0000000000 RBX: 0000000000000000 RCX: ffffffff8797fd9d
> > > RDX: 0000000000000019 RSI: ffffffff8797fde6 RDI: 00000000000000c8
> > > RBP: ffff888099848040 R08: ffff88809a5f6440 R09: fffffbfff1860b4c
> > > R10: ffffffff8c305a5f R11: fffffbfff1860b4b R12: ffff88809984857e
> > > R13: 0000000000000000 R14: ffff888086aa4000 R15: 0000000000000000
> > > FS:  00000000009b4880(0000) GS:ffff8880ae700000(0000) knlGS:0000000000000000
> > > CS:  0010 DS: 0000 ES: 0000 CR0: 0000000080050033
> > > CR2: 0000000020000140 CR3: 00000000a7fdf000 CR4: 00000000001406e0
> > > DR0: 0000000000000000 DR1: 0000000000000000 DR2: 0000000000000000
> > > DR3: 0000000000000000 DR6: 00000000fffe0ff0 DR7: 0000000000000400
> > > Call Trace:
> > >  tipc_sendstream+0x4c/0x70 net/tipc/socket.c:1533
> > >  sock_sendmsg_nosec net/socket.c:652 [inline]
> > >  sock_sendmsg+0xcf/0x120 net/socket.c:672
> > >  ____sys_sendmsg+0x32f/0x810 net/socket.c:2352
> > >  ___sys_sendmsg+0x100/0x170 net/socket.c:2406
> > >  __sys_sendmmsg+0x195/0x480 net/socket.c:2496
> > >  __do_sys_sendmmsg net/socket.c:2525 [inline]
> > >  __se_sys_sendmmsg net/socket.c:2522 [inline]
> > >  __x64_sys_sendmmsg+0x99/0x100 net/socket.c:2522
> > >  do_syscall_64+0xf6/0x7d0 arch/x86/entry/common.c:295
> > >  entry_SYSCALL_64_after_hwframe+0x49/0xb3
> > > RIP: 0033:0x440199
> > > ...
> > >
> > > This bug was bisected to commit 0a3e060f340d ("tipc: add test for Nagle
> > > algorithm effectiveness"). However, it is not the case, the trouble was
> > > from the base in the case of zero data length message sending, we would
> > > unexpectedly make an empty 'txq' queue after the 'tipc_msg_append()' in
> > > Nagle mode.
> > >
> > > A similar crash can be generated even without the bisected patch but at
> > > the link layer when it accesses the empty queue.
> > >
> > > We solve the issues by building at least one buffer to go with socket's
> > > header and an optional data section that may be empty like what we had
> > > with the 'tipc_msg_build()'.
> > >
> > > Note: the previous commit 4c21daae3dbc ("tipc: Fix NULL pointer
> > > dereference in __tipc_sendstream()") is obsoleted by this one since the
> > > 'txq' will be never empty and the check of 'skb != NULL' is unnecessary
> > > but it is safe anyway.
> > Hi, Tuong
> >
> > If commit 4c21daae3dbc is obsoleted by this one, can you please
> > send a patch to revert it?
> >
> > Thanks.
> Hi Xin,
>
> That patch includes a sanity check which is always true and safe, so I don’t think
> we need to revert it. Do you agree?
surely it's safe.

People may be confused when reading the code:
                                if (skb) {
                                        msg_set_ack_required(buf_msg(skb));
                                        tsk->expect_ack = true;
                                } else {
                                        tsk->expect_ack = false;  <----- [1]
                                }

like why expect_ack needs to be set to false in [1]

>
> BR/Tuong
> >
> > >
> > > Reported-by: syz...@sy...
> > > Fixes: c0bceb97db9e ("tipc: add smart nagle feature")
> > > Acked-by: Jon Maloy <jm...@re...>
> > > Signed-off-by: Tuong Lien <tuo...@de...>
> > > ---
> > >  net/tipc/msg.c | 4 ++--
> > >  1 file changed, 2 insertions(+), 2 deletions(-)
> > >
> > > diff --git a/net/tipc/msg.c b/net/tipc/msg.c
> > > index c0afcd627c5e..046e4cb3acea 100644
> > > --- a/net/tipc/msg.c
> > > +++ b/net/tipc/msg.c
> > > @@ -221,7 +221,7 @@ int tipc_msg_append(struct tipc_msg *_hdr, struct msghdr *m, int dlen,
> > >         accounted = skb ? msg_blocks(buf_msg(skb)) : 0;
> > >         total = accounted;
> > >
> > > -       while (rem) {
> > > +       do {
> > >                 if (!skb || skb->len >= mss) {
> > >                         skb = tipc_buf_acquire(mss, GFP_KERNEL);
> > >                         if (unlikely(!skb))
> > > @@ -245,7 +245,7 @@ int tipc_msg_append(struct tipc_msg *_hdr, struct msghdr *m, int dlen,
> > >                 skb_put(skb, cpy);
> > >                 rem -= cpy;
> > >                 total += msg_blocks(hdr) - curr;
> > > -       }
> > > +       } while (rem);
> > >         return total - accounted;
> > >  }
> > >
> > > --
> > > 2.13.7
> > >
> > >
> > >
> > > _______________________________________________
> > > tipc-discussion mailing list
> > > tip...@li...
> > > https://lists.sourceforge.net/lists/listinfo/tipc-discussion

Re: [tipc-discussion] [net-next] tipc: fix NULL pointer dereference in streaming

[Tuong T. L. <tuo...@de...>]

> -----Original Message-----
> From: Xin Long <luc...@gm...>
> Sent: Tuesday, July 21, 2020 6:23 PM
> To: Tuong Tong Lien <tuo...@de...>
> Cc: davem <da...@da...>; jm...@re...; ma...@do...; Ying Xue <yin...@wi...>; network dev
> <ne...@vg...>; tip...@li...
> Subject: Re: [tipc-discussion] [net-next] tipc: fix NULL pointer dereference in streaming
> 
> On Wed, Jun 3, 2020 at 1:06 PM Tuong Lien <tuo...@de...> wrote:
> >
> > syzbot found the following crash:
> >
> > general protection fault, probably for non-canonical address 0xdffffc0000000019: 0000 [#1] PREEMPT SMP KASAN
> > KASAN: null-ptr-deref in range [0x00000000000000c8-0x00000000000000cf]
> > CPU: 1 PID: 7060 Comm: syz-executor394 Not tainted 5.7.0-rc6-syzkaller #0
> > Hardware name: Google Google Compute Engine/Google Compute Engine, BIOS Google 01/01/2011
> > RIP: 0010:__tipc_sendstream+0xbde/0x11f0 net/tipc/socket.c:1591
> > Code: 00 00 00 00 48 39 5c 24 28 48 0f 44 d8 e8 fa 3e db f9 48 b8 00 00 00 00 00 fc ff df 48 8d bb c8 00 00 00 48 89 fa 48 c1 ea 03 <80> 3c
> 02 00 0f 85 e2 04 00 00 48 8b 9b c8 00 00 00 48 b8 00 00 00
> > RSP: 0018:ffffc90003ef7818 EFLAGS: 00010202
> > RAX: dffffc0000000000 RBX: 0000000000000000 RCX: ffffffff8797fd9d
> > RDX: 0000000000000019 RSI: ffffffff8797fde6 RDI: 00000000000000c8
> > RBP: ffff888099848040 R08: ffff88809a5f6440 R09: fffffbfff1860b4c
> > R10: ffffffff8c305a5f R11: fffffbfff1860b4b R12: ffff88809984857e
> > R13: 0000000000000000 R14: ffff888086aa4000 R15: 0000000000000000
> > FS:  00000000009b4880(0000) GS:ffff8880ae700000(0000) knlGS:0000000000000000
> > CS:  0010 DS: 0000 ES: 0000 CR0: 0000000080050033
> > CR2: 0000000020000140 CR3: 00000000a7fdf000 CR4: 00000000001406e0
> > DR0: 0000000000000000 DR1: 0000000000000000 DR2: 0000000000000000
> > DR3: 0000000000000000 DR6: 00000000fffe0ff0 DR7: 0000000000000400
> > Call Trace:
> >  tipc_sendstream+0x4c/0x70 net/tipc/socket.c:1533
> >  sock_sendmsg_nosec net/socket.c:652 [inline]
> >  sock_sendmsg+0xcf/0x120 net/socket.c:672
> >  ____sys_sendmsg+0x32f/0x810 net/socket.c:2352
> >  ___sys_sendmsg+0x100/0x170 net/socket.c:2406
> >  __sys_sendmmsg+0x195/0x480 net/socket.c:2496
> >  __do_sys_sendmmsg net/socket.c:2525 [inline]
> >  __se_sys_sendmmsg net/socket.c:2522 [inline]
> >  __x64_sys_sendmmsg+0x99/0x100 net/socket.c:2522
> >  do_syscall_64+0xf6/0x7d0 arch/x86/entry/common.c:295
> >  entry_SYSCALL_64_after_hwframe+0x49/0xb3
> > RIP: 0033:0x440199
> > ...
> >
> > This bug was bisected to commit 0a3e060f340d ("tipc: add test for Nagle
> > algorithm effectiveness"). However, it is not the case, the trouble was
> > from the base in the case of zero data length message sending, we would
> > unexpectedly make an empty 'txq' queue after the 'tipc_msg_append()' in
> > Nagle mode.
> >
> > A similar crash can be generated even without the bisected patch but at
> > the link layer when it accesses the empty queue.
> >
> > We solve the issues by building at least one buffer to go with socket's
> > header and an optional data section that may be empty like what we had
> > with the 'tipc_msg_build()'.
> >
> > Note: the previous commit 4c21daae3dbc ("tipc: Fix NULL pointer
> > dereference in __tipc_sendstream()") is obsoleted by this one since the
> > 'txq' will be never empty and the check of 'skb != NULL' is unnecessary
> > but it is safe anyway.
> Hi, Tuong
> 
> If commit 4c21daae3dbc is obsoleted by this one, can you please
> send a patch to revert it?
> 
> Thanks.
Hi Xin,

That patch includes a sanity check which is always true and safe, so I don’t think
we need to revert it. Do you agree?

BR/Tuong
> 
> >
> > Reported-by: syz...@sy...
> > Fixes: c0bceb97db9e ("tipc: add smart nagle feature")
> > Acked-by: Jon Maloy <jm...@re...>
> > Signed-off-by: Tuong Lien <tuo...@de...>
> > ---
> >  net/tipc/msg.c | 4 ++--
> >  1 file changed, 2 insertions(+), 2 deletions(-)
> >
> > diff --git a/net/tipc/msg.c b/net/tipc/msg.c
> > index c0afcd627c5e..046e4cb3acea 100644
> > --- a/net/tipc/msg.c
> > +++ b/net/tipc/msg.c
> > @@ -221,7 +221,7 @@ int tipc_msg_append(struct tipc_msg *_hdr, struct msghdr *m, int dlen,
> >         accounted = skb ? msg_blocks(buf_msg(skb)) : 0;
> >         total = accounted;
> >
> > -       while (rem) {
> > +       do {
> >                 if (!skb || skb->len >= mss) {
> >                         skb = tipc_buf_acquire(mss, GFP_KERNEL);
> >                         if (unlikely(!skb))
> > @@ -245,7 +245,7 @@ int tipc_msg_append(struct tipc_msg *_hdr, struct msghdr *m, int dlen,
> >                 skb_put(skb, cpy);
> >                 rem -= cpy;
> >                 total += msg_blocks(hdr) - curr;
> > -       }
> > +       } while (rem);
> >         return total - accounted;
> >  }
> >
> > --
> > 2.13.7
> >
> >
> >
> > _______________________________________________
> > tipc-discussion mailing list
> > tip...@li...
> > https://lists.sourceforge.net/lists/listinfo/tipc-discussion

Re: [tipc-discussion] [net-next] tipc: fix NULL pointer dereference in streaming

[Xin L. <luc...@gm...>]

On Wed, Jun 3, 2020 at 1:06 PM Tuong Lien <tuo...@de...> wrote:
>
> syzbot found the following crash:
>
> general protection fault, probably for non-canonical address 0xdffffc0000000019: 0000 [#1] PREEMPT SMP KASAN
> KASAN: null-ptr-deref in range [0x00000000000000c8-0x00000000000000cf]
> CPU: 1 PID: 7060 Comm: syz-executor394 Not tainted 5.7.0-rc6-syzkaller #0
> Hardware name: Google Google Compute Engine/Google Compute Engine, BIOS Google 01/01/2011
> RIP: 0010:__tipc_sendstream+0xbde/0x11f0 net/tipc/socket.c:1591
> Code: 00 00 00 00 48 39 5c 24 28 48 0f 44 d8 e8 fa 3e db f9 48 b8 00 00 00 00 00 fc ff df 48 8d bb c8 00 00 00 48 89 fa 48 c1 ea 03 <80> 3c 02 00 0f 85 e2 04 00 00 48 8b 9b c8 00 00 00 48 b8 00 00 00
> RSP: 0018:ffffc90003ef7818 EFLAGS: 00010202
> RAX: dffffc0000000000 RBX: 0000000000000000 RCX: ffffffff8797fd9d
> RDX: 0000000000000019 RSI: ffffffff8797fde6 RDI: 00000000000000c8
> RBP: ffff888099848040 R08: ffff88809a5f6440 R09: fffffbfff1860b4c
> R10: ffffffff8c305a5f R11: fffffbfff1860b4b R12: ffff88809984857e
> R13: 0000000000000000 R14: ffff888086aa4000 R15: 0000000000000000
> FS:  00000000009b4880(0000) GS:ffff8880ae700000(0000) knlGS:0000000000000000
> CS:  0010 DS: 0000 ES: 0000 CR0: 0000000080050033
> CR2: 0000000020000140 CR3: 00000000a7fdf000 CR4: 00000000001406e0
> DR0: 0000000000000000 DR1: 0000000000000000 DR2: 0000000000000000
> DR3: 0000000000000000 DR6: 00000000fffe0ff0 DR7: 0000000000000400
> Call Trace:
>  tipc_sendstream+0x4c/0x70 net/tipc/socket.c:1533
>  sock_sendmsg_nosec net/socket.c:652 [inline]
>  sock_sendmsg+0xcf/0x120 net/socket.c:672
>  ____sys_sendmsg+0x32f/0x810 net/socket.c:2352
>  ___sys_sendmsg+0x100/0x170 net/socket.c:2406
>  __sys_sendmmsg+0x195/0x480 net/socket.c:2496
>  __do_sys_sendmmsg net/socket.c:2525 [inline]
>  __se_sys_sendmmsg net/socket.c:2522 [inline]
>  __x64_sys_sendmmsg+0x99/0x100 net/socket.c:2522
>  do_syscall_64+0xf6/0x7d0 arch/x86/entry/common.c:295
>  entry_SYSCALL_64_after_hwframe+0x49/0xb3
> RIP: 0033:0x440199
> ...
>
> This bug was bisected to commit 0a3e060f340d ("tipc: add test for Nagle
> algorithm effectiveness"). However, it is not the case, the trouble was
> from the base in the case of zero data length message sending, we would
> unexpectedly make an empty 'txq' queue after the 'tipc_msg_append()' in
> Nagle mode.
>
> A similar crash can be generated even without the bisected patch but at
> the link layer when it accesses the empty queue.
>
> We solve the issues by building at least one buffer to go with socket's
> header and an optional data section that may be empty like what we had
> with the 'tipc_msg_build()'.
>
> Note: the previous commit 4c21daae3dbc ("tipc: Fix NULL pointer
> dereference in __tipc_sendstream()") is obsoleted by this one since the
> 'txq' will be never empty and the check of 'skb != NULL' is unnecessary
> but it is safe anyway.
Hi, Tuong

If commit 4c21daae3dbc is obsoleted by this one, can you please
send a patch to revert it?

Thanks.

>
> Reported-by: syz...@sy...
> Fixes: c0bceb97db9e ("tipc: add smart nagle feature")
> Acked-by: Jon Maloy <jm...@re...>
> Signed-off-by: Tuong Lien <tuo...@de...>
> ---
>  net/tipc/msg.c | 4 ++--
>  1 file changed, 2 insertions(+), 2 deletions(-)
>
> diff --git a/net/tipc/msg.c b/net/tipc/msg.c
> index c0afcd627c5e..046e4cb3acea 100644
> --- a/net/tipc/msg.c
> +++ b/net/tipc/msg.c
> @@ -221,7 +221,7 @@ int tipc_msg_append(struct tipc_msg *_hdr, struct msghdr *m, int dlen,
>         accounted = skb ? msg_blocks(buf_msg(skb)) : 0;
>         total = accounted;
>
> -       while (rem) {
> +       do {
>                 if (!skb || skb->len >= mss) {
>                         skb = tipc_buf_acquire(mss, GFP_KERNEL);
>                         if (unlikely(!skb))
> @@ -245,7 +245,7 @@ int tipc_msg_append(struct tipc_msg *_hdr, struct msghdr *m, int dlen,
>                 skb_put(skb, cpy);
>                 rem -= cpy;
>                 total += msg_blocks(hdr) - curr;
> -       }
> +       } while (rem);
>         return total - accounted;
>  }
>
> --
> 2.13.7
>
>
>
> _______________________________________________
> tipc-discussion mailing list
> tip...@li...
> https://lists.sourceforge.net/lists/listinfo/tipc-discussion

Re: [tipc-discussion] [PATCH RFC 3/5] tipc: introduce encryption master key

[Tuong T. L. <tuo...@de...>]

Hi Jon,

Thanks a lot for your comments, I will update them accordingly!
Regarding the 'master key' definition, I have just done a search and found it to perfectly matched our case:

--------------
https://en.wikipedia.org/wiki/Cryptographic_key_types
Symmetric master key
    A symmetric master key is used to derive other symmetric keys (e.g., data encryption keys, key wrapping keys, or authentication keys) using symmetric cryptographic methods.
--------------

I think it is fine to go with it, of course unless we find a better one... 😊!

BR/Tuong

-----Original Message-----
From: Jon Maloy <jm...@re...> 
Sent: Tuesday, July 21, 2020 1:07 AM
To: Tuong Tong Lien <tuo...@de...>; ma...@do...; yin...@wi...; tip...@li...
Cc: tipc-dek <tip...@de...>
Subject: Re: [PATCH RFC 3/5] tipc: introduce encryption master key



On 7/10/20 6:11 AM, Tuong Lien wrote:
> In addition to the supported cluster & per-node encryption keys for the
> en/decryption of TIPC messages, we now introduce one option for user to
> set a cluster key as 'master key',
Still not quite happy with the term 'master key' but have no really good 
suggestions at the moment.
Maybe 'membership key' or just 'member key'?
> which is simply a symmetric key like
> the former
but has a longer life cycle. It has two purposes:

  - Authentication of new member nodes in the cluster.
   New nodes, having no knowledge of current session keys
   in the cluster will still be able to join the cluster as long as
   they know this key. This is because all neighbor discovery
   (LINK_CONFIG) messages must be encrypted with this key.

- Encryption of session encryption keys during automatic exchange
   and update of those. This is  a feature we will introduce
   in a later commit in this series.
We insert the new key into the currently unused slot 0 in the key array,
and start using it immediately once the user has set it.
>
After joining, a node only knowing the master key should be fully
> communicable to existing nodes in the cluster, although those nodes may
> have their own session keys activated (i.e. not the master one). To
> support this, we simply
s/simply//
> define a 'grace period', starting from the time
> a node itself reports having no RX keys, so the existing nodes will use
> the master key for encryption instead. The grace period can be extended
> but will automatically stop after, e.g. 5 seconds without a new report.
> This is also the basis for later key exchanging since the new node will
> be impossible to decrypt anything without the support from master key.
>
> For user to set a master key, we define a new netlink flag -
> 'TIPC_NLA_NODE_KEY_MASTER', so it can be added to the current 'set key'
> netlink command to specify the setting key to be a master key.
>
> Above all, the traditional cluster/per-node key mechanism is guaranteed
> to work when user comes not to use this master key option. This is also
> compatible to legacy nodes without the feature supported.
Even this new key can be updated without any interruption
of cluster connectivity, but is so is needed, this has to be
coordinated and set by the user.
>
> Signed-off-by: Tuong Lien <tuo...@de...>
> ---
>   include/uapi/linux/tipc_netlink.h |   1 +
>   net/tipc/crypto.c                 | 206 ++++++++++++++++++++++++++++----------
>   net/tipc/crypto.h                 |  15 ++-
>   net/tipc/msg.h                    |   4 +-
>   net/tipc/netlink.c                |   1 +
>   net/tipc/node.c                   |  46 ++++-----
>   6 files changed, 189 insertions(+), 84 deletions(-)
>
> diff --git a/include/uapi/linux/tipc_netlink.h b/include/uapi/linux/tipc_netlink.h
> index dc0d23a50e69..d484baa9d365 100644
> --- a/include/uapi/linux/tipc_netlink.h
> +++ b/include/uapi/linux/tipc_netlink.h
> @@ -165,6 +165,7 @@ enum {
>   	TIPC_NLA_NODE_UP,		/* flag */
>   	TIPC_NLA_NODE_ID,		/* data */
>   	TIPC_NLA_NODE_KEY,		/* data */
> +	TIPC_NLA_NODE_KEY_MASTER,	/* flag */
>   
>   	__TIPC_NLA_NODE_MAX,
>   	TIPC_NLA_NODE_MAX = __TIPC_NLA_NODE_MAX - 1
> diff --git a/net/tipc/crypto.c b/net/tipc/crypto.c
> index f1046ab4bd01..7c688cd0b13e 100644
> --- a/net/tipc/crypto.c
> +++ b/net/tipc/crypto.c
> @@ -38,6 +38,7 @@
>   #include <crypto/aes.h>
>   #include "crypto.h"
>   
> +#define TIPC_TX_GRACE_PERIOD	msecs_to_jiffies(5000) /* 5s */
>   #define TIPC_TX_LASTING_TIME	msecs_to_jiffies(10000) /* 10s */
>   #define TIPC_RX_ACTIVE_LIM	msecs_to_jiffies(3000) /* 3s */
>   #define TIPC_RX_PASSIVE_LIM	msecs_to_jiffies(15000) /* 15s */
> @@ -49,9 +50,9 @@
>    * TIPC Key ids
>    */
>   enum {
> -	KEY_UNUSED = 0,
> -	KEY_MIN,
> -	KEY_1 = KEY_MIN,
> +	KEY_MASTER = 0,
> +	KEY_MIN = KEY_MASTER,
> +	KEY_1 = 1,
>   	KEY_2,
>   	KEY_3,
>   	KEY_MAX = KEY_3,
> @@ -166,27 +167,36 @@ struct tipc_crypto_stats {
>    * @aead: array of pointers to AEAD keys for encryption/decryption
>    * @peer_rx_active: replicated peer RX active key index
>    * @key: the key states
> - * @working: the crypto is working or not
>    * @stats: the crypto statistics
>    * @name: the crypto name
>    * @sndnxt: the per-peer sndnxt (TX)
>    * @timer1: general timer 1 (jiffies)
>    * @timer2: general timer 2 (jiffies)
> + * @working: the crypto is working or not
> + * @key_master: flag indicates if master key exists
> + * @legacy_user: flag indicates if a peer joins w/o master key (for bwd comp.)
>    * @lock: tipc_key lock
>    */
>   struct tipc_crypto {
>   	struct net *net;
>   	struct tipc_node *node;
> -	struct tipc_aead __rcu *aead[KEY_MAX + 1]; /* key[0] is UNUSED */
> +	struct tipc_aead __rcu *aead[KEY_MAX + 1];
>   	atomic_t peer_rx_active;
>   	struct tipc_key key;
> -	u8 working:1;
>   	struct tipc_crypto_stats __percpu *stats;
>   	char name[48];
>   
>   	atomic64_t sndnxt ____cacheline_aligned;
>   	unsigned long timer1;
>   	unsigned long timer2;
> +	union {
> +		struct {
> +			u8 working:1;
> +			u8 key_master:1;
> +			u8 legacy_user:1;
> +		};
> +		u8 flags;
> +	};
>   	spinlock_t lock; /* crypto lock */
>   
>   } ____cacheline_aligned;
> @@ -236,13 +246,19 @@ static inline void tipc_crypto_key_set_state(struct tipc_crypto *c,
>   					     u8 new_active,
>   					     u8 new_pending);
>   static int tipc_crypto_key_attach(struct tipc_crypto *c,
> -				  struct tipc_aead *aead, u8 pos);
> +				  struct tipc_aead *aead, u8 pos,
> +				  bool master_key);
>   static bool tipc_crypto_key_try_align(struct tipc_crypto *rx, u8 new_pending);
>   static struct tipc_aead *tipc_crypto_key_pick_tx(struct tipc_crypto *tx,
>   						 struct tipc_crypto *rx,
> -						 struct sk_buff *skb);
> +						 struct sk_buff *skb,
> +						 u8 tx_key);
>   static void tipc_crypto_key_synch(struct tipc_crypto *rx, struct sk_buff *skb);
>   static int tipc_crypto_key_revoke(struct net *net, u8 tx_key);
> +static inline void tipc_crypto_clone_msg(struct net *net, struct sk_buff *_skb,
> +					 struct tipc_bearer *b,
> +					 struct tipc_media_addr *dst,
> +					 struct tipc_node *__dnode, u8 type);
>   static void tipc_crypto_rcv_complete(struct net *net, struct tipc_aead *aead,
>   				     struct tipc_bearer *b,
>   				     struct sk_buff **skb, int err);
> @@ -937,8 +953,6 @@ bool tipc_ehdr_validate(struct sk_buff *skb)
>   		return false;
>   	if (unlikely(skb->len <= ehsz + TIPC_AES_GCM_TAG_SIZE))
>   		return false;
> -	if (unlikely(!ehdr->tx_key))
> -		return false;
>   
>   	return true;
>   }
> @@ -991,6 +1005,8 @@ static int tipc_ehdr_build(struct net *net, struct tipc_aead *aead,
>   	ehdr->tx_key = tx_key;
>   	ehdr->destined = (__rx) ? 1 : 0;
>   	ehdr->rx_key_active = (__rx) ? __rx->key.active : 0;
> +	ehdr->rx_nokey = (__rx) ? !__rx->key.keys : 0;
> +	ehdr->master_key = aead->crypto->key_master;
>   	ehdr->reserved_1 = 0;
>   	ehdr->reserved_2 = 0;
>   
> @@ -1033,6 +1049,7 @@ static inline void tipc_crypto_key_set_state(struct tipc_crypto *c,
>    * @c: TIPC crypto to which new key is attached
>    * @ukey: the user key
>    * @mode: the key mode (CLUSTER_KEY or PER_NODE_KEY)
> + * @master_key: specify this is a cluster master key
>    *
>    * A new TIPC AEAD key will be allocated and initiated with the specified user
>    * key, then attached to the TIPC crypto.
> @@ -1040,7 +1057,7 @@ static inline void tipc_crypto_key_set_state(struct tipc_crypto *c,
>    * Return: new key id in case of success, otherwise: < 0
>    */
>   int tipc_crypto_key_init(struct tipc_crypto *c, struct tipc_aead_key *ukey,
> -			 u8 mode)
> +			 u8 mode, bool master_key)
>   {
>   	struct tipc_aead *aead = NULL;
>   	int rc = 0;
> @@ -1053,7 +1070,7 @@ int tipc_crypto_key_init(struct tipc_crypto *c, struct tipc_aead_key *ukey,
>   	}
>   
>   	/* Attach it to the crypto */
> -	rc = tipc_crypto_key_attach(c, aead, 0);
> +	rc = tipc_crypto_key_attach(c, aead, 0, master_key);
>   	if (rc < 0) {
>   		pr_err("%s: unable to attach key, err %d\n", c->name, rc);
>   		tipc_aead_free(&aead->rcu);
> @@ -1069,11 +1086,13 @@ int tipc_crypto_key_init(struct tipc_crypto *c, struct tipc_aead_key *ukey,
>    * @c: TIPC crypto to which the new AEAD key is attached
>    * @aead: the new AEAD key pointer
>    * @pos: desired slot in the crypto key array, = 0 if any!
> + * @master_key: specify this is a cluster master key
>    *
>    * Return: new key id in case of success, otherwise: -EBUSY
>    */
>   static int tipc_crypto_key_attach(struct tipc_crypto *c,
> -				  struct tipc_aead *aead, u8 pos)
> +				  struct tipc_aead *aead, u8 pos,
> +				  bool master_key)
>   {
>   	struct tipc_key key;
>   	int rc = -EBUSY;
> @@ -1081,6 +1100,10 @@ static int tipc_crypto_key_attach(struct tipc_crypto *c,
>   
>   	spin_lock_bh(&c->lock);
>   	key = c->key;
> +	if (master_key) {
> +		new_key = KEY_MASTER;
> +		goto attach;
> +	}
>   	if (key.active && key.passive)
>   		goto exit;
>   	if (key.pending) {
> @@ -1112,8 +1135,7 @@ static int tipc_crypto_key_attach(struct tipc_crypto *c,
>   		tipc_crypto_key_set_state(c, key.passive, key.active,
>   					  key.pending);
>   	c->working = 1;
> -	c->timer1 = jiffies;
> -	c->timer2 = jiffies;
> +	c->key_master |= master_key;
>   	rc = new_key;
>   
>   exit:
> @@ -1126,7 +1148,7 @@ void tipc_crypto_key_flush(struct tipc_crypto *c)
>   	int k;
>   
>   	spin_lock_bh(&c->lock);
> -	c->working = 0;
> +	c->flags = 0;
>   	tipc_crypto_key_set_state(c, 0, 0, 0);
>   	for (k = KEY_MIN; k <= KEY_MAX; k++)
>   		tipc_crypto_key_detach(c->aead[k], &c->lock);
> @@ -1202,6 +1224,7 @@ static bool tipc_crypto_key_try_align(struct tipc_crypto *rx, u8 new_pending)
>    * @tx: TX crypto handle
>    * @rx: RX crypto handle (can be NULL)
>    * @skb: the message skb which will be decrypted later
> + * @tx_key: peer TX key id
>    *
>    * This function looks up the existing TX keys and pick one which is suitable
>    * for the message decryption, that must be a cluster key and not used before
> @@ -1211,7 +1234,8 @@ static bool tipc_crypto_key_try_align(struct tipc_crypto *rx, u8 new_pending)
>    */
>   static struct tipc_aead *tipc_crypto_key_pick_tx(struct tipc_crypto *tx,
>   						 struct tipc_crypto *rx,
> -						 struct sk_buff *skb)
> +						 struct sk_buff *skb,
> +						 u8 tx_key)
>   {
>   	struct tipc_skb_cb *skb_cb = TIPC_SKB_CB(skb);
>   	struct tipc_aead *aead = NULL;
> @@ -1230,6 +1254,10 @@ static struct tipc_aead *tipc_crypto_key_pick_tx(struct tipc_crypto *tx,
>   
>   	/* Pick one TX key */
>   	spin_lock(&tx->lock);
> +	if (tx_key == KEY_MASTER) {
> +		aead = tipc_aead_rcu_ptr(tx->aead[KEY_MASTER], &tx->lock);
> +		goto done;
> +	}
>   	do {
>   		k = (i == 0) ? key.pending :
>   			((i == 1) ? key.active : key.passive);
> @@ -1249,9 +1277,12 @@ static struct tipc_aead *tipc_crypto_key_pick_tx(struct tipc_crypto *tx,
>   		skb->next = skb_clone(skb, GFP_ATOMIC);
>   		if (unlikely(!skb->next))
>   			pr_warn("Failed to clone skb for next round if any\n");
> -		WARN_ON(!refcount_inc_not_zero(&aead->refcnt));
>   		break;
>   	} while (++i < 3);
> +
> +done:
> +	if (likely(aead))
> +		WARN_ON(!refcount_inc_not_zero(&aead->refcnt));
>   	spin_unlock(&tx->lock);
>   
>   	return aead;
> @@ -1266,6 +1297,9 @@ static struct tipc_aead *tipc_crypto_key_pick_tx(struct tipc_crypto *tx,
>    * has changed, so the number of TX keys' users on this node are increased and
>    * decreased correspondingly.
>    *
> + * It also considers if peer has no key, then we need to make own master key
> + * (if any) taking over i.e. starting grace period.
> + *
>    * The "per-peer" sndnxt is also reset when the peer key has switched.
>    */
>   static void tipc_crypto_key_synch(struct tipc_crypto *rx, struct sk_buff *skb)
> @@ -1276,11 +1310,23 @@ static void tipc_crypto_key_synch(struct tipc_crypto *rx, struct sk_buff *skb)
>   	u32 self = tipc_own_addr(rx->net);
>   	u8 cur, new;
>   
> -	/* Ensure this message is destined to us first */
> +	/* Update RX 'key_master' flag according to peer, also mark "legacy" if
> +	 * a peer has no master key.
> +	 */
> +	rx->key_master = ehdr->master_key;
> +	if (!rx->key_master)
> +		tx->legacy_user = 1;
> +
> +	/* For later cases, apply only if message is destined to this node */
>   	if (!ehdr->destined || msg_short(hdr) || msg_destnode(hdr) != self)
>   		return;
>   
> -	/* Peer RX active key has changed, let's update own TX users */
> +	/* Case 1: Peer has no keys, let's make master key taking over */
s/taking/take/
> +	if (ehdr->rx_nokey)
> +		/* Set or extend grace period */
> +		tx->timer2 = jiffies;
> +
> +	/* Case 2: Peer RX active key has changed, let's update own TX users */
>   	cur = atomic_read(&rx->peer_rx_active);
>   	new = ehdr->rx_key_active;
>   	if (tx->key.keys &&
> @@ -1338,7 +1384,7 @@ int tipc_crypto_start(struct tipc_crypto **crypto, struct net *net,
>   		return -ENOMEM;
>   	}
>   
> -	c->working = 0;
> +	c->flags = 0;
>   	c->net = net;
>   	c->node = node;
>   	tipc_crypto_key_set_state(c, 0, 0, 0);
> @@ -1473,6 +1519,12 @@ void tipc_crypto_timeout(struct tipc_crypto *rx)
>   s5:
>   	spin_unlock(&rx->lock);
>   
> +	/* Relax it here, the flag will be set again if it really is, but only
> +	 * when we are not in grace period for safety!
> +	 */
> +	if (time_after(jiffies, tx->timer2 + TIPC_TX_GRACE_PERIOD))
> +		tx->legacy_user = 0;
> +
>   	/* Limit max_tfms & do debug commands if needed */
>   	if (likely(sysctl_tipc_max_tfms <= TIPC_MAX_TFMS_LIM))
>   		return;
> @@ -1482,6 +1534,22 @@ void tipc_crypto_timeout(struct tipc_crypto *rx)
>   	tipc_crypto_do_cmd(rx->net, cmd);
>   }
>   
> +static inline void tipc_crypto_clone_msg(struct net *net, struct sk_buff *_skb,
> +					 struct tipc_bearer *b,
> +					 struct tipc_media_addr *dst,
> +					 struct tipc_node *__dnode, u8 type)
> +{
> +	struct sk_buff *skb;
> +
> +	skb = skb_clone(_skb, GFP_ATOMIC);
> +	if (skb) {
> +		TIPC_SKB_CB(skb)->xmit_type = type;
> +		tipc_crypto_xmit(net, &skb, b, dst, __dnode);
> +		if (skb)
> +			b->media->send_msg(net, skb, b, dst);
> +	}
> +}
> +
>   /**
>    * tipc_crypto_xmit - Build & encrypt TIPC message for xmit
>    * @net: struct net
> @@ -1491,7 +1559,8 @@ void tipc_crypto_timeout(struct tipc_crypto *rx)
>    * @__dnode: destination node for reference if any
>    *
>    * First, build an encryption message header on the top of the message, then
> - * encrypt the original TIPC message by using the active or pending TX key.
> + * encrypt the original TIPC message by using the pending, master or active
> + * key with this preference order.
>    * If the encryption is successful, the encrypted skb is returned directly or
>    * via the callback.
>    * Otherwise, the skb is freed!
> @@ -1514,46 +1583,63 @@ int tipc_crypto_xmit(struct net *net, struct sk_buff **skb,
>   	struct tipc_msg *hdr = buf_msg(*skb);
>   	struct tipc_key key = tx->key;
>   	struct tipc_aead *aead = NULL;
> -	struct sk_buff *_skb;
> -	int rc = -ENOKEY;
>   	u32 user = msg_user(hdr);
> -	u8 tx_key;
> +	u32 type = msg_type(hdr);
> +	int rc = -ENOKEY;
> +	u8 tx_key = 0;
>   
>   	/* No encryption? */
>   	if (!tx->working)
>   		return 0;
>   
> -	/* Try with the pending key if available and:
> -	 * 1) This is the only choice (i.e. no active key) or;
> -	 * 2) Peer has switched to this key (unicast only) or;
> -	 * 3) It is time to do a pending key probe;
> -	 */
> +	/* Pending key if peer has active on it or probing time */
>   	if (unlikely(key.pending)) {
>   		tx_key = key.pending;
> -		if (!key.active)
> +		if (!tx->key_master && !key.active)
>   			goto encrypt;
>   		if (__rx && atomic_read(&__rx->peer_rx_active) == tx_key)
>   			goto encrypt;
> -		if (TIPC_SKB_CB(*skb)->probe) {
> +		if (TIPC_SKB_CB(*skb)->xmit_type == SKB_PROBING) {
>   			pr_debug("%s: probing for key[%d]\n", tx->name,
>   				 key.pending);
>   			goto encrypt;
>   		}
> -		if (user == LINK_CONFIG || user == LINK_PROTOCOL) {
> -			_skb = skb_clone(*skb, GFP_ATOMIC);
> -			if (_skb) {
> -				TIPC_SKB_CB(_skb)->probe = 1;
> -				tipc_crypto_xmit(net, &_skb, b, dst, __dnode);
> -				if (_skb)
> -					b->media->send_msg(net, _skb, b, dst);
> +		if (user == LINK_CONFIG || user == LINK_PROTOCOL)
> +			tipc_crypto_clone_msg(net, *skb, b, dst, __dnode,
> +					      SKB_PROBING);
> +	}
> +
> +	/* Master key if this is a *vital* message or in grace period */
> +	if (tx->key_master) {
> +		tx_key = KEY_MASTER;
> +		if (!key.active)
> +			goto encrypt;
> +		if (TIPC_SKB_CB(*skb)->xmit_type == SKB_GRACING) {
> +			pr_debug("%s: gracing for msg (%d %d)\n", tx->name,
> +				 user, type);
> +			goto encrypt;
> +		}
> +		if (user == LINK_CONFIG ||
> +		    (user == LINK_PROTOCOL && type == RESET_MSG) ||
> +		    time_before(jiffies, tx->timer2 + TIPC_TX_GRACE_PERIOD)) {
Why do RESET messages need to be encrypted with the master key?
> +			if (__rx && __rx->key_master &&
> +			    !atomic_read(&__rx->peer_rx_active))
> +				goto encrypt;
> +			if (!__rx) {
> +				if (likely(!tx->legacy_user))
> +					goto encrypt;
> +				tipc_crypto_clone_msg(net, *skb, b, dst,
> +						      __dnode, SKB_GRACING);
>   			}
>   		}
>   	}
> +
>   	/* Else, use the active key if any */
>   	if (likely(key.active)) {
>   		tx_key = key.active;
>   		goto encrypt;
>   	}
> +
>   	goto exit;
>   
>   encrypt:
> @@ -1619,15 +1705,16 @@ int tipc_crypto_rcv(struct net *net, struct tipc_crypto *rx,
>   	struct tipc_aead *aead = NULL;
>   	struct tipc_key key;
>   	int rc = -ENOKEY;
> -	u8 tx_key = 0;
> +	u8 tx_key;
> +
> +	tx_key = ((struct tipc_ehdr *)(*skb)->data)->tx_key;
>   
>   	/* New peer?
>   	 * Let's try with TX key (i.e. cluster mode) & verify the skb first!
>   	 */
> -	if (unlikely(!rx))
> +	if (unlikely(!rx || tx_key == KEY_MASTER))
>   		goto pick_tx;
>   
> -	tx_key = ((struct tipc_ehdr *)(*skb)->data)->tx_key;
>   	/* Pick RX key according to TX key if any */
>   	key = rx->key;
>   	if (tx_key == key.active || tx_key == key.pending ||
> @@ -1640,7 +1727,7 @@ int tipc_crypto_rcv(struct net *net, struct tipc_crypto *rx,
>   
>   pick_tx:
>   	/* No key suitable? Try to pick one from TX... */
> -	aead = tipc_crypto_key_pick_tx(tx, rx, *skb);
> +	aead = tipc_crypto_key_pick_tx(tx, rx, *skb, tx_key);
>   	if (aead)
>   		goto decrypt;
>   	goto exit;
> @@ -1722,9 +1809,12 @@ static void tipc_crypto_rcv_complete(struct net *net, struct tipc_aead *aead,
>   				goto free_skb;
>   		}
>   
> +		/* Ignore cloning if it was TX master key */
> +		if (ehdr->tx_key == KEY_MASTER)
> +			goto rcv;
>   		if (tipc_aead_clone(&tmp, aead) < 0)
>   			goto rcv;
> -		if (tipc_crypto_key_attach(rx, tmp, ehdr->tx_key) < 0) {
> +		if (tipc_crypto_key_attach(rx, tmp, ehdr->tx_key, false) < 0) {
>   			tipc_aead_free(&tmp->rcu);
>   			goto rcv;
>   		}
> @@ -1740,10 +1830,10 @@ static void tipc_crypto_rcv_complete(struct net *net, struct tipc_aead *aead,
>   	/* Set the RX key's user */
>   	tipc_aead_users_set(aead, 1);
>   
> -rcv:
>   	/* Mark this point, RX works */
>   	rx->timer1 = jiffies;
>   
> +rcv:
>   	/* Remove ehdr & auth. tag prior to tipc_rcv() */
>   	ehdr = (struct tipc_ehdr *)(*skb)->data;
>   
> @@ -1865,14 +1955,24 @@ static char *tipc_crypto_key_dump(struct tipc_crypto *c, char *buf)
>   	char *s;
>   
>   	for (k = KEY_MIN; k <= KEY_MAX; k++) {
> -		if (k == key.passive)
> -			s = "PAS";
> -		else if (k == key.active)
> -			s = "ACT";
> -		else if (k == key.pending)
> -			s = "PEN";
> -		else
> -			s = "-";
> +		if (k == KEY_MASTER) {
> +			if (is_rx(c))
> +				continue;
> +			if (time_before(jiffies,
> +					c->timer2 + TIPC_TX_GRACE_PERIOD))
> +				s = "ACT";
> +			else
> +				s = "PAS";
> +		} else {
> +			if (k == key.passive)
> +				s = "PAS";
> +			else if (k == key.active)
> +				s = "ACT";
> +			else if (k == key.pending)
> +				s = "PEN";
> +			else
> +				s = "-";
> +		}
>   		i += scnprintf(buf + i, 200 - i, "\tKey%d: %s", k, s);
>   
>   		rcu_read_lock();
> diff --git a/net/tipc/crypto.h b/net/tipc/crypto.h
> index c3de769f49e8..7fcb80cb0e8a 100644
> --- a/net/tipc/crypto.h
> +++ b/net/tipc/crypto.h
> @@ -74,7 +74,7 @@ extern int sysctl_tipc_max_tfms __read_mostly;
>    *     3 3 2 2 2 2 2 2 2 2 2 2 1 1 1 1 1 1 1 1 1 1 0 0 0 0 0 0 0 0 0 0
>    *     1 0 9 8 7 6 5 4|3 2 1 0 9 8 7 6|5 4 3 2 1 0 9 8|7 6 5 4 3 2 1 0
>    *    +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
> - * w0:|Ver=7| User  |D|TX |RX |K|                 Rsvd                |
> + * w0:|Ver=7| User  |D|TX |RX |K|M|N|             Rsvd                |
>    *    +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
>    * w1:|                             Seqno                             |
>    * w2:|                           (8 octets)                          |
> @@ -101,6 +101,9 @@ extern int sysctl_tipc_max_tfms __read_mostly;
>    *	RX	: Currently RX active key corresponding to the destination
>    *	          node's TX key (when the "D" bit is set)
>    *	K	: Keep-alive bit (for RPS, LINK_PROTOCOL/STATE_MSG only)
> + *	M       : Bit indicates if sender has master key
> + *	N	: Bit indicates if sender has no RX keys corresponding to the
> + *	          receiver's TX (when the "D" bit is set)
>    *	Rsvd	: Reserved bit, field
>    * Word1-2:
>    *	Seqno	: The 64-bit sequence number of the encrypted message, also
> @@ -117,7 +120,9 @@ struct tipc_ehdr {
>   			__u8	destined:1,
>   				user:4,
>   				version:3;
> -			__u8	reserved_1:3,
> +			__u8	reserved_1:1,
> +				rx_nokey:1,
> +				master_key:1,
>   				keepalive:1,
>   				rx_key_active:2,
>   				tx_key:2;
> @@ -128,7 +133,9 @@ struct tipc_ehdr {
>   			__u8	tx_key:2,
>   				rx_key_active:2,
>   				keepalive:1,
> -				reserved_1:3;
> +				master_key:1,
> +				rx_nokey:1,
> +				reserved_1:1;
>   #else
>   #error  "Please fix <asm/byteorder.h>"
>   #endif
> @@ -158,7 +165,7 @@ int tipc_crypto_xmit(struct net *net, struct sk_buff **skb,
>   int tipc_crypto_rcv(struct net *net, struct tipc_crypto *rx,
>   		    struct sk_buff **skb, struct tipc_bearer *b);
>   int tipc_crypto_key_init(struct tipc_crypto *c, struct tipc_aead_key *ukey,
> -			 u8 mode);
> +			 u8 mode, bool master_key);
>   void tipc_crypto_key_flush(struct tipc_crypto *c);
>   int tipc_aead_key_validate(struct tipc_aead_key *ukey);
>   bool tipc_ehdr_validate(struct sk_buff *skb);
> diff --git a/net/tipc/msg.h b/net/tipc/msg.h
> index 1016e96db5c4..25e5c5c8a6ff 100644
> --- a/net/tipc/msg.h
> +++ b/net/tipc/msg.h
> @@ -127,7 +127,9 @@ struct tipc_skb_cb {
>   #ifdef CONFIG_TIPC_CRYPTO
>   			u8 encrypted:1;
>   			u8 decrypted:1;
> -			u8 probe:1;
> +#define SKB_PROBING	1
> +#define SKB_GRACING	2
> +			u8 xmit_type:2;
>   			u8 tx_clone_deferred:1;
>   #endif
>   		};
> diff --git a/net/tipc/netlink.c b/net/tipc/netlink.c
> index c4aee6247d55..1ec00fcc26ee 100644
> --- a/net/tipc/netlink.c
> +++ b/net/tipc/netlink.c
> @@ -108,6 +108,7 @@ const struct nla_policy tipc_nl_node_policy[TIPC_NLA_NODE_MAX + 1] = {
>   					    .len = TIPC_NODEID_LEN},
>   	[TIPC_NLA_NODE_KEY]		= { .type = NLA_BINARY,
>   					    .len = TIPC_AEAD_KEY_SIZE_MAX},
> +	[TIPC_NLA_NODE_KEY_MASTER]	= { .type = NLA_FLAG },
>   };
>   
>   /* Properties valid for media, bearer and link */
> diff --git a/net/tipc/node.c b/net/tipc/node.c
> index 030a51c4d1fa..55f012d1ea74 100644
> --- a/net/tipc/node.c
> +++ b/net/tipc/node.c
> @@ -2872,11 +2872,11 @@ static int __tipc_nl_node_set_key(struct sk_buff *skb, struct genl_info *info)
>   {
>   	struct nlattr *attrs[TIPC_NLA_NODE_MAX + 1];
>   	struct net *net = sock_net(skb->sk);
> -	struct tipc_net *tn = tipc_net(net);
> +	struct tipc_crypto *tx = tipc_net(net)->crypto_tx, *c = tx;
>   	struct tipc_node *n = NULL;
>   	struct tipc_aead_key *ukey;
> -	struct tipc_crypto *c;
> -	u8 *id, *own_id;
> +	bool master_key = false;
> +	u8 *id, *own_id, mode;
>   	int rc = 0;
>   
>   	if (!info->attrs[TIPC_NLA_NODE])
> @@ -2886,51 +2886,45 @@ static int __tipc_nl_node_set_key(struct sk_buff *skb, struct genl_info *info)
>   			      info->attrs[TIPC_NLA_NODE],
>   			      tipc_nl_node_policy, info->extack);
>   	if (rc)
> -		goto exit;
> +		return rc;
>   
>   	own_id = tipc_own_id(net);
> -	if (!own_id) {
> -		rc = -EPERM;
> -		goto exit;
> -	}
> +	if (!own_id)
> +		return -EPERM;
>   
>   	rc = tipc_nl_retrieve_key(attrs, &ukey);
>   	if (rc)
> -		goto exit;
> +		return rc;
>   
>   	rc = tipc_aead_key_validate(ukey);
>   	if (rc)
> -		goto exit;
> +		return rc;
>   
>   	rc = tipc_nl_retrieve_nodeid(attrs, &id);
>   	switch (rc) {
>   	case -ENODATA:
> -		/* Cluster key mode */
> -		rc = tipc_crypto_key_init(tn->crypto_tx, ukey, CLUSTER_KEY);
> +		mode = CLUSTER_KEY;
> +		master_key = !!(attrs[TIPC_NLA_NODE_KEY_MASTER]);
>   		break;
>   	case 0:
> -		/* Per-node key mode */
> -		if (!memcmp(id, own_id, NODE_ID_LEN)) {
> -			c = tn->crypto_tx;
> -		} else {
> +		mode = PER_NODE_KEY;
> +		if (memcmp(id, own_id, NODE_ID_LEN)) {
>   			n = tipc_node_find_by_id(net, id) ?:
>   				tipc_node_create(net, 0, id, 0xffffu, 0, true);
> -			if (unlikely(!n)) {
> -				rc = -ENOMEM;
> -				break;
> -			}
> +			if (unlikely(!n))
> +				return -ENOMEM;
>   			c = n->crypto_rx;
>   		}
> -
> -		rc = tipc_crypto_key_init(c, ukey, PER_NODE_KEY);
> -		if (n)
> -			tipc_node_put(n);
>   		break;
>   	default:
> -		break;
> +		return rc;
>   	}
>   
> -exit:
> +	/* Initiate the TX/RX key */
> +	rc = tipc_crypto_key_init(c, ukey, mode, master_key);
> +	if (n)
> +		tipc_node_put(n);
> +
>   	return (rc < 0) ? rc : 0;
>   }
>   
Acked-by: Jon Maloy <jm...@re...>

Re: [tipc-discussion] [net v1 1/1] tipc: allow to build NACK message in link timeout function

[David M. <da...@da...>]

From: Tung Nguyen <tun...@de...>
Date: Tue, 21 Jul 2020 08:57:05 +0700

> Commit 02288248b051 ("tipc: eliminate gap indicator from ACK messages")
> eliminated sending of the 'gap' indicator in regular ACK messages and
> only allowed to build NACK message with enabled probe/probe_reply.
> However, necessary correction for building NACK message was missed
> in tipc_link_timeout() function. This leads to significant delay and
> link reset (due to retransmission failure) in lossy environment.
> 
> This commit fixes it by setting the 'probe' flag to 'true' when
> the receive deferred queue is not empty. As a result, NACK message
> will be built to send back to another peer.
> 
> Fixes: commit 02288248b051 ("tipc: eliminate gap indicator from ACK messages")

"commit" doesn't belong here, I removed it when I applied your patch.

> Acked-by: Jon Maloy <jm...@re...>
> Signed-off-by: Tung Nguyen <tun...@de...>

Applied and queued up for -stable, thanks.

[tipc-discussion] [net v1 1/1] tipc: allow to build NACK message in link timeout function

[Tung N. <tun...@de...>]

Commit 02288248b051 ("tipc: eliminate gap indicator from ACK messages")
eliminated sending of the 'gap' indicator in regular ACK messages and
only allowed to build NACK message with enabled probe/probe_reply.
However, necessary correction for building NACK message was missed
in tipc_link_timeout() function. This leads to significant delay and
link reset (due to retransmission failure) in lossy environment.

This commit fixes it by setting the 'probe' flag to 'true' when
the receive deferred queue is not empty. As a result, NACK message
will be built to send back to another peer.

Fixes: commit 02288248b051 ("tipc: eliminate gap indicator from ACK messages")
Acked-by: Jon Maloy <jm...@re...>
Signed-off-by: Tung Nguyen <tun...@de...>
---
 net/tipc/link.c | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/net/tipc/link.c b/net/tipc/link.c
index 263d950e70e9..d40f8e5b7683 100644
--- a/net/tipc/link.c
+++ b/net/tipc/link.c
@@ -827,11 +827,11 @@ int tipc_link_timeout(struct tipc_link *l, struct sk_buff_head *xmitq)
 		state |= l->bc_rcvlink->rcv_unacked;
 		state |= l->rcv_unacked;
 		state |= !skb_queue_empty(&l->transmq);
-		state |= !skb_queue_empty(&l->deferdq);
 		probe = mstate->probing;
 		probe |= l->silent_intv_cnt;
 		if (probe || mstate->monitoring)
 			l->silent_intv_cnt++;
+		probe |= !skb_queue_empty(&l->deferdq);
 		if (l->snd_nxt == l->checkpoint) {
 			tipc_link_update_cwin(l, 0, 0);
 			probe = true;
-- 
2.17.1

Re: [tipc-discussion] [PATCH RFC 5/5] tipc: add automatic rekeying for encryption key

[Jon M. <jm...@re...>]

On 7/10/20 6:11 AM, Tuong Lien wrote:
> Rekeying is required for security since a key is less secure when using
> for a long time. Also, key will be detached when its nonce value (or
> seqno ...) is exhausted. We now make the rekeying process automatic and
> configurable by user.
>
> Basically, TIPC will at a specific interval generate a new key by using
> the kernel 'Random Number Generator' cipher, then attach it as the node
> TX key and securely distribute to others in the cluster as RX keys (-
> the key exchange). The automatic key switching will then take over, and
> make the new key active shortly. Afterwards, the traffic from this node
> will be encrypted with the new session key. The same can happen in peer
> nodes but not necessarily at the same time.
>
> For simplicity, the automatically generated key will be initiated as a
> per node key. It is not too hard to also support a cluster key rekeying
> (e.g. a given node will generate a unique cluster key and update to the
> others in the cluster...), but that doesn't bring much benefit, while a
> per-node key is even more security.
s/security/secure/
>
> We also enable user to force a rekeying or change the rekeying interval
> via netlink, the new 'set key' command option: 'TIPC_NLA_NODE_REKEYING'
> is added for these purposes as follows:
> - A value >= 1 will be set as the rekeying interval (in minutes);
> - A value of 0 will disable the rekeying;
> - A value of 'TIPC_REKEYING_NOW' (~0) will force an immediate rekeying;
>
> The default rekeying interval is (60 * 24) minutes i.e. done every day.
> There isn't any restriction for the value but user shouldn't set it too
> small or too large which results in an "ineffective" rekeying (thats ok
> for testing though).
>
> Signed-off-by: Tuong Lien <tuo...@de...>
> ---
>   include/uapi/linux/tipc.h         |   2 +
>   include/uapi/linux/tipc_netlink.h |   1 +
>   net/tipc/crypto.c                 | 119 +++++++++++++++++++++++++++++++++++++-
>   net/tipc/crypto.h                 |   2 +
>   net/tipc/netlink.c                |   1 +
>   net/tipc/node.c                   |  28 ++++++++-
>   6 files changed, 148 insertions(+), 5 deletions(-)
>
> diff --git a/include/uapi/linux/tipc.h b/include/uapi/linux/tipc.h
> index add01db1daef..80ea15e12113 100644
> --- a/include/uapi/linux/tipc.h
> +++ b/include/uapi/linux/tipc.h
> @@ -254,6 +254,8 @@ static inline int tipc_aead_key_size(struct tipc_aead_key *key)
>   	return sizeof(*key) + key->keylen;
>   }
>   
> +#define TIPC_REKEYING_NOW		(~0U)
> +
>   /* The macros and functions below are deprecated:
>    */
>   
> diff --git a/include/uapi/linux/tipc_netlink.h b/include/uapi/linux/tipc_netlink.h
> index d484baa9d365..d847dd671d79 100644
> --- a/include/uapi/linux/tipc_netlink.h
> +++ b/include/uapi/linux/tipc_netlink.h
> @@ -166,6 +166,7 @@ enum {
>   	TIPC_NLA_NODE_ID,		/* data */
>   	TIPC_NLA_NODE_KEY,		/* data */
>   	TIPC_NLA_NODE_KEY_MASTER,	/* flag */
> +	TIPC_NLA_NODE_REKEYING,		/* u32 */
>   
>   	__TIPC_NLA_NODE_MAX,
>   	TIPC_NLA_NODE_MAX = __TIPC_NLA_NODE_MAX - 1
> diff --git a/net/tipc/crypto.c b/net/tipc/crypto.c
> index 9d00555ede83..d497585010ee 100644
> --- a/net/tipc/crypto.c
> +++ b/net/tipc/crypto.c
> @@ -36,6 +36,7 @@
>   
>   #include <crypto/aead.h>
>   #include <crypto/aes.h>
> +#include <crypto/rng.h>
>   #include "crypto.h"
>   #include "msg.h"
>   #include "bcast.h"
> @@ -48,6 +49,8 @@
>   #define TIPC_MAX_TFMS_DEF	10
>   #define TIPC_MAX_TFMS_LIM	1000
>   
> +#define TIPC_REKEYING_INTV_DEF	(60 * 24) /* default: 1 day */
> +
>   /**
>    * TIPC Key ids
>    */
> @@ -181,6 +184,7 @@ struct tipc_crypto_stats {
>    * @wq: common workqueue on TX crypto
>    * @work: delayed work sched for TX/RX
>    * @key_distr: key distributing state
> + * @rekeying_intv: rekeying interval (in minutes)
>    * @stats: the crypto statistics
>    * @name: the crypto name
>    * @sndnxt: the per-peer sndnxt (TX)
> @@ -206,6 +210,7 @@ struct tipc_crypto {
>   #define KEY_DISTR_SCHED		1
>   #define KEY_DISTR_COMPL		2
>   	atomic_t key_distr;
> +	u32 rekeying_intv;
>   
>   	struct tipc_crypto_stats __percpu *stats;
>   	char name[48];
> @@ -294,7 +299,9 @@ static char *tipc_key_change_dump(struct tipc_key old, struct tipc_key new,
>   static int tipc_crypto_key_xmit(struct net *net, struct tipc_aead_key *skey,
>   				u16 gen, u8 mode, u32 dnode);
>   static bool tipc_crypto_key_rcv(struct tipc_crypto *rx, struct tipc_msg *hdr);
> +static void tipc_crypto_work_tx(struct work_struct *work);
>   static void tipc_crypto_work_rx(struct work_struct *work);
> +static int tipc_aead_key_generate(struct tipc_aead_key *skey);
>   
>   #define is_tx(crypto) (!(crypto)->node)
>   #define is_rx(crypto) (!is_tx(crypto))
> @@ -342,6 +349,27 @@ int tipc_aead_key_validate(struct tipc_aead_key *ukey)
>   	return 0;
>   }
>   
> +/**
> + * tipc_aead_key_generate - Generate new session key
> + * @skey: input/output key with new content
> + *
> + * Return: 0 in case of success, otherwise < 0
> + */
> +static int tipc_aead_key_generate(struct tipc_aead_key *skey)
> +{
> +	int rc = 0;
> +
> +	/* Fill the key's content with a random value via RNG cipher */
> +	rc = crypto_get_default_rng();
> +	if (likely(!rc)) {
> +		rc = crypto_rng_get_bytes(crypto_default_rng, skey->key,
> +					  skey->keylen);
> +		crypto_put_default_rng();
> +	}
> +
> +	return rc;
> +}
> +
>   static struct tipc_aead *tipc_aead_get(struct tipc_aead __rcu *aead)
>   {
>   	struct tipc_aead *tmp;
> @@ -1471,13 +1499,16 @@ int tipc_crypto_start(struct tipc_crypto **crypto, struct net *net,
>   	atomic64_set(&c->sndnxt, 0);
>   	c->timer1 = jiffies;
>   	c->timer2 = jiffies;
> +	c->rekeying_intv = TIPC_REKEYING_INTV_DEF;
>   	spin_lock_init(&c->lock);
>   	scnprintf(c->name, 48, "%s(%s)", (is_rx(c)) ? "RX" : "TX",
>   		  (is_rx(c)) ? tipc_node_get_id_str(c->node) :
>   			       tipc_own_id_string(c->net));
>   
> -        if (is_rx(c))
> -                INIT_DELAYED_WORK(&c->work, tipc_crypto_work_rx);
> +	if (is_rx(c))
> +		INIT_DELAYED_WORK(&c->work, tipc_crypto_work_rx);
> +	else
> +		INIT_DELAYED_WORK(&c->work, tipc_crypto_work_tx);
>   
>   	*crypto = c;
>   	return 0;
> @@ -1492,8 +1523,11 @@ void tipc_crypto_stop(struct tipc_crypto **crypto)
>   		return;
>   
>   	/* Flush any queued works & destroy wq */
> -	if (is_tx(c))
> +	if (is_tx(c)) {
> +		c->rekeying_intv = 0;
> +		cancel_delayed_work_sync(&c->work);
>   		destroy_workqueue(c->wq);
> +	}
>   
>   	/* Release AEAD keys */
>   	rcu_read_lock();
> @@ -2346,3 +2380,82 @@ static void tipc_crypto_work_rx(struct work_struct *work)
>   
>   	tipc_node_put(rx->node);
>   }
> +
> +/**
> + * tipc_crypto_rekeying_sched - (Re)schedule rekeying w/o new interval
> + * @tx: TX crypto
> + * @changed: if the rekeying needs to be rescheduled with new interval
> + * @new_intv: new rekeying interval (when "changed" = true)
> + */
> +void tipc_crypto_rekeying_sched(struct tipc_crypto *tx, bool changed,
> +				u32 new_intv)
> +{
> +	unsigned long delay;
> +	bool now = false;
> +
> +	if (changed) {
> +		if (new_intv == TIPC_REKEYING_NOW)
> +			now = true;
> +		else
> +			tx->rekeying_intv = new_intv;
> +		cancel_delayed_work_sync(&tx->work);
> +	}
> +
> +	if (tx->rekeying_intv || now) {
> +		delay = (now) ? 0 : tx->rekeying_intv * 60 * 1000;
> +		queue_delayed_work(tx->wq, &tx->work, msecs_to_jiffies(delay));
> +	}
> +}
> +
> +/**
> + * tipc_crypto_work_tx - Scheduled TX works handler
> + * @work: the struct TX work
> + *
> + * The function processes the previous scheduled work, i.e. key rekeying, by
> + * generating a new session key based on current one, then attaching it to the
> + * TX crypto and finally distributing it to peers. It also re-schedules the
> + * rekeying if needed.
> + */
> +static void tipc_crypto_work_tx(struct work_struct *work)
> +{
> +	struct delayed_work *dwork = to_delayed_work(work);
> +	struct tipc_crypto *tx = container_of(dwork, struct tipc_crypto, work);
> +	struct tipc_aead_key *skey = NULL;
> +	struct tipc_key key = tx->key;
> +	struct tipc_aead *aead;
> +	int rc = -ENOMEM;
> +
> +	if (unlikely(key.pending))
> +		goto resched;
> +
> +	/* Take current key as a template */
> +	rcu_read_lock();
> +	aead = rcu_dereference(tx->aead[key.active ?: KEY_MASTER]);
> +	if (unlikely(!aead)) {
> +		rcu_read_unlock();
> +		/* At least one key should exist for securing */
> +		return;
> +	}
> +
> +	/* Lets duplicate it first */
> +	skey = kmemdup(aead->key, tipc_aead_key_size(aead->key), GFP_ATOMIC);
> +	rcu_read_unlock();
> +
> +	/* Now, generate new key, initiate & distribute it */
> +	if (likely(skey)) {
> +		rc = tipc_aead_key_generate(skey) ?:
> +		     tipc_crypto_key_init(tx, skey, PER_NODE_KEY, false);
> +		if (likely(rc > 0))
> +			rc = tipc_crypto_key_distr(tx, rc, NULL);
> +		kzfree(skey);
> +	}
> +
> +	if (likely(!rc))
> +		pr_info("%s: rekeying has been done\n", tx->name);
> +	else
> +		pr_warn_ratelimited("%s: rekeying returns %d\n", tx->name, rc);
> +
> +resched:
> +	/* Re-schedule rekeying if any */
> +	tipc_crypto_rekeying_sched(tx, false, 0);
> +}
> diff --git a/net/tipc/crypto.h b/net/tipc/crypto.h
> index 70bda3d7e174..e1f4e8fb5c10 100644
> --- a/net/tipc/crypto.h
> +++ b/net/tipc/crypto.h
> @@ -171,6 +171,8 @@ void tipc_crypto_key_flush(struct tipc_crypto *c);
>   int tipc_crypto_key_distr(struct tipc_crypto *tx, u8 key,
>   			  struct tipc_node *dest);
>   void tipc_crypto_msg_rcv(struct net *net, struct sk_buff *skb);
> +void tipc_crypto_rekeying_sched(struct tipc_crypto *tx, bool changed,
> +				u32 new_intv);
>   int tipc_aead_key_validate(struct tipc_aead_key *ukey);
>   bool tipc_ehdr_validate(struct sk_buff *skb);
>   
> diff --git a/net/tipc/netlink.c b/net/tipc/netlink.c
> index 1ec00fcc26ee..c447cb5f879e 100644
> --- a/net/tipc/netlink.c
> +++ b/net/tipc/netlink.c
> @@ -109,6 +109,7 @@ const struct nla_policy tipc_nl_node_policy[TIPC_NLA_NODE_MAX + 1] = {
>   	[TIPC_NLA_NODE_KEY]		= { .type = NLA_BINARY,
>   					    .len = TIPC_AEAD_KEY_SIZE_MAX},
>   	[TIPC_NLA_NODE_KEY_MASTER]	= { .type = NLA_FLAG },
> +	[TIPC_NLA_NODE_REKEYING]	= { .type = NLA_U32 },
>   };
>   
>   /* Properties valid for media, bearer and link */
> diff --git a/net/tipc/node.c b/net/tipc/node.c
> index 186951b690d2..95f045cf9f55 100644
> --- a/net/tipc/node.c
> +++ b/net/tipc/node.c
> @@ -2877,6 +2877,17 @@ static int tipc_nl_retrieve_nodeid(struct nlattr **attrs, u8 **node_id)
>   	return 0;
>   }
>   
> +static int tipc_nl_retrieve_rekeying(struct nlattr **attrs, u32 *intv)
> +{
> +	struct nlattr *attr = attrs[TIPC_NLA_NODE_REKEYING];
> +
> +	if (!attr)
> +		return -ENODATA;
> +
> +	*intv = nla_get_u32(attr);
> +	return 0;
> +}
> +
>   static int __tipc_nl_node_set_key(struct sk_buff *skb, struct genl_info *info)
>   {
>   	struct nlattr *attrs[TIPC_NLA_NODE_MAX + 1];
> @@ -2884,8 +2895,9 @@ static int __tipc_nl_node_set_key(struct sk_buff *skb, struct genl_info *info)
>   	struct tipc_crypto *tx = tipc_net(net)->crypto_tx, *c = tx;
>   	struct tipc_node *n = NULL;
>   	struct tipc_aead_key *ukey;
> -	bool master_key = false;
> +	bool rekeying = true, master_key = false;
>   	u8 *id, *own_id, mode;
> +	u32 intv = 0;
>   	int rc = 0;
>   
>   	if (!info->attrs[TIPC_NLA_NODE])
> @@ -2901,9 +2913,17 @@ static int __tipc_nl_node_set_key(struct sk_buff *skb, struct genl_info *info)
>   	if (!own_id)
>   		return -EPERM;
>   
> +	rc = tipc_nl_retrieve_rekeying(attrs, &intv);
> +	if (rc == -ENODATA)
> +		rekeying = false;
> +
>   	rc = tipc_nl_retrieve_key(attrs, &ukey);
> -	if (rc)
> +	if (rc == -ENODATA && rekeying) {
> +		rc = 0;
> +		goto rekeying;
> +	} else if (rc) {
>   		return rc;
> +	}
>   
>   	rc = tipc_aead_key_validate(ukey);
>   	if (rc)
> @@ -2938,6 +2958,10 @@ static int __tipc_nl_node_set_key(struct sk_buff *skb, struct genl_info *info)
>   	if (!master_key)
>   		tipc_crypto_key_distr(tx, rc, NULL);
>   
> +rekeying:
> +	/* Schedule TX rekeying if needed */
> +	tipc_crypto_rekeying_sched(tx, rekeying, intv);
> +
>   exit:
>   	if (n)
>   		tipc_node_put(n);

As usual a nice job from you.

Acked-by: Jon Maloy <jm...@re...>

Re: [tipc-discussion] [PATCH RFC 4/5] tipc: add automatic session key exchange

[Jon M. <jm...@re...>]

On 7/10/20 6:11 AM, Tuong Lien wrote:
> With support from the master key option in the previous commit,
it becomes easy to make frequent updates/exchanges of session
keys between authenticated cluster nodes.

> Basically, there are two situations where the key exchange will
> take in place:
>
> - When a new node joins the cluster (with the master key), it will need
>    to get its peer's TX key, so that be able to decrypt further messages
>    from that peer.
>
> - When a new session key is generated (by either user manual setting or
>    later automatic rekeying feature), the key will be distributed to all
>    peer nodes in the cluster.
>
> A key to be exchanged is encapsulated in the data part of a 'MSG_CRYPTO
> /KEY_DISTR_MSG' TIPC v2 message, then xmit-ed as usual and encrypted by
> using the master key before sending out. Upon receipt of the message it
> will be decrypted in the same way as regular messages, then attached as
> the sender's RX key in the receiver node.
>
> In this way, the key exchange is reliable by the link layer, as well as
> security, integrity and authenticity by the crypto layer.
>
> Also, the forward security will be easily achieved by user changing the
> master key actively but
this should not be required very frequently.

The key exchange feature is dependent on the presence of a master key,
is still optional. It can be turned off/on via
sysfs: 'net/tipc/key_exchange_enabled' (default 1: enabled).
Note however that the master key still is needed for new nodes to be able to join the cluster.


You should mention something about compatibility here. Does this allow a seamless upgrade in a cluster where we have encryption but the master key is not supported?


  Signed-off-by: Tuong Lien <tuo...@de...>
---
  net/tipc/crypto.c | 359 +++++++++++++++++++++++++++++++++++++++++++++++++++---
  net/tipc/crypto.h |  24 ++++
  net/tipc/link.c   |   5 +
  net/tipc/msg.h    |   6 +
  net/tipc/node.c   |  19 ++-
  net/tipc/node.h   |   2 +
  net/tipc/sysctl.c |   9 ++
  7 files changed, 404 insertions(+), 20 deletions(-)

diff --git a/net/tipc/crypto.c b/net/tipc/crypto.c
index 7c688cd0b13e..9d00555ede83 100644
--- a/net/tipc/crypto.c
+++ b/net/tipc/crypto.c
@@ -37,6 +37,8 @@
  #include <crypto/aead.h>
  #include <crypto/aes.h>
  #include "crypto.h"
+#include "msg.h"
+#include "bcast.h"
  
  #define TIPC_TX_GRACE_PERIOD	msecs_to_jiffies(5000) /* 5s */
  #define TIPC_TX_LASTING_TIME	msecs_to_jiffies(10000) /* 10s */
@@ -82,6 +84,8 @@ static const char *hstats[MAX_STATS] = {"ok", "nok", "async", "async_ok",
  
  /* Max TFMs number per key */
  int sysctl_tipc_max_tfms __read_mostly = TIPC_MAX_TFMS_DEF;
+/* Key exchange switch, default: on */
+int sysctl_tipc_key_exchange_enabled __read_mostly = 1;
  
  /**
   * struct tipc_key - TIPC keys' status indicator
@@ -133,6 +137,8 @@ struct tipc_tfm {
   * @mode: crypto mode is applied to the key
   * @hint[]: a hint for user key
   * @rcu: struct rcu_head
+ * @key: the aead key
+ * @gen: the key's generation
   * @seqno: the key seqno (cluster scope)
   * @refcnt: the key reference counter
   */
@@ -147,6 +153,8 @@ struct tipc_aead {
  	u8 mode;
  	char hint[2 * TIPC_AEAD_HINT_LEN + 1];
  	struct rcu_head rcu;
+	struct tipc_aead_key *key;
+	u16 gen;
  
  	atomic64_t seqno ____cacheline_aligned;
  	refcount_t refcnt ____cacheline_aligned;
@@ -166,7 +174,13 @@ struct tipc_crypto_stats {
   * @node: TIPC node (RX)
   * @aead: array of pointers to AEAD keys for encryption/decryption
   * @peer_rx_active: replicated peer RX active key index
+ * @key_gen: TX/RX key generation
   * @key: the key states
+ * @skey_mode: session key's mode
+ * @skey: received session key
+ * @wq: common workqueue on TX crypto
+ * @work: delayed work sched for TX/RX
+ * @key_distr: key distributing state
   * @stats: the crypto statistics
   * @name: the crypto name
   * @sndnxt: the per-peer sndnxt (TX)
@@ -175,6 +189,7 @@ struct tipc_crypto_stats {
   * @working: the crypto is working or not
   * @key_master: flag indicates if master key exists
   * @legacy_user: flag indicates if a peer joins w/o master key (for bwd comp.)
+ * @nokey: no key indication
   * @lock: tipc_key lock
   */
  struct tipc_crypto {
@@ -182,7 +197,16 @@ struct tipc_crypto {
  	struct tipc_node *node;
  	struct tipc_aead __rcu *aead[KEY_MAX + 1];
  	atomic_t peer_rx_active;
+	u16 key_gen;
  	struct tipc_key key;
+	u8 skey_mode;
+	struct tipc_aead_key *skey;
+	struct workqueue_struct *wq;
+	struct delayed_work work;
+#define KEY_DISTR_SCHED		1
+#define KEY_DISTR_COMPL		2
+	atomic_t key_distr;
+
  	struct tipc_crypto_stats __percpu *stats;
  	char name[48];
  
@@ -194,6 +218,7 @@ struct tipc_crypto {
  			u8 working:1;
  			u8 key_master:1;
  			u8 legacy_user:1;
+			u8 nokey: 1;
  		};
  		u8 flags;
  	};
@@ -266,6 +291,11 @@ static void tipc_crypto_do_cmd(struct net *net, int cmd);
  static char *tipc_crypto_key_dump(struct tipc_crypto *c, char *buf);
  static char *tipc_key_change_dump(struct tipc_key old, struct tipc_key new,
  				  char *buf);
+static int tipc_crypto_key_xmit(struct net *net, struct tipc_aead_key *skey,
+				u16 gen, u8 mode, u32 dnode);
+static bool tipc_crypto_key_rcv(struct tipc_crypto *rx, struct tipc_msg *hdr);
+static void tipc_crypto_work_rx(struct work_struct *work);
+
  #define is_tx(crypto) (!(crypto)->node)
  #define is_rx(crypto) (!is_tx(crypto))
  
@@ -356,6 +386,7 @@ static void tipc_aead_free(struct rcu_head *rp)
  		kfree(head);
  	}
  	free_percpu(aead->tfm_entry);
+	kzfree(aead->key);
  	kfree(aead);
  }
  
@@ -526,6 +557,7 @@ static int tipc_aead_init(struct tipc_aead **aead, struct tipc_aead_key *ukey,
  	tmp->mode = mode;
  	tmp->cloned = NULL;
  	tmp->authsize = TIPC_AES_GCM_TAG_SIZE;
+	tmp->key = kmemdup(ukey, tipc_aead_key_size(ukey), GFP_KERNEL);
  	memcpy(&tmp->salt, ukey->key + keylen, TIPC_AES_GCM_SALT_SIZE);
  	atomic_set(&tmp->users, 0);
  	atomic64_set(&tmp->seqno, 0);
@@ -1005,7 +1037,7 @@ static int tipc_ehdr_build(struct net *net, struct tipc_aead *aead,
  	ehdr->tx_key = tx_key;
  	ehdr->destined = (__rx) ? 1 : 0;
  	ehdr->rx_key_active = (__rx) ? __rx->key.active : 0;
-	ehdr->rx_nokey = (__rx) ? !__rx->key.keys : 0;
+	ehdr->rx_nokey = (__rx) ? __rx->nokey : 0;
  	ehdr->master_key = aead->crypto->key_master;
  	ehdr->reserved_1 = 0;
  	ehdr->reserved_2 = 0;
@@ -1130,11 +1162,13 @@ static int tipc_crypto_key_attach(struct tipc_crypto *c,
  
  attach:
  	aead->crypto = c;
+	aead->gen = (is_tx(c)) ? ++c->key_gen : c->key_gen;
  	tipc_aead_rcu_replace(c->aead[new_key], aead, &c->lock);
  	if (likely(c->key.keys != key.keys))
  		tipc_crypto_key_set_state(c, key.passive, key.active,
  					  key.pending);
  	c->working = 1;
+	c->nokey = 0;
  	c->key_master |= master_key;
  	rc = new_key;
  
@@ -1145,14 +1179,33 @@ static int tipc_crypto_key_attach(struct tipc_crypto *c,
  
  void tipc_crypto_key_flush(struct tipc_crypto *c)
  {
+	struct tipc_crypto *tx, *rx;
  	int k;
  
  	spin_lock_bh(&c->lock);
+	if (is_rx(c)) {
+		/* Try to cancel pending work */
+		rx = c;
+		tx = tipc_net(rx->net)->crypto_tx;
+		if (cancel_delayed_work(&rx->work)) {
+			kfree(rx->skey);
+			rx->skey = NULL;
+			atomic_xchg(&rx->key_distr, 0);
+			tipc_node_put(rx->node);
+		}
+		/* RX stopping => decrease TX key users if any */
+		k = atomic_xchg(&rx->peer_rx_active, 0);
+		if (k) {
+			tipc_aead_users_dec(tx->aead[k], 0);
+			/* Mark the point TX key users changed */
+			tx->timer1 = jiffies;
+		}
+	}
+
  	c->flags = 0;
  	tipc_crypto_key_set_state(c, 0, 0, 0);
  	for (k = KEY_MIN; k <= KEY_MAX; k++)
  		tipc_crypto_key_detach(c->aead[k], &c->lock);
-	atomic_set(&c->peer_rx_active, 0);
  	atomic64_set(&c->sndnxt, 0);
  	spin_unlock_bh(&c->lock);
  }
@@ -1298,7 +1351,8 @@ static struct tipc_aead *tipc_crypto_key_pick_tx(struct tipc_crypto *tx,
   * decreased correspondingly.
   *
   * It also considers if peer has no key, then we need to make own master key
- * (if any) taking over i.e. starting grace period.
+ * (if any) taking over i.e. starting grace period and also trigger key
+ * distributing process.
   *
   * The "per-peer" sndnxt is also reset when the peer key has switched.
   */
@@ -1309,6 +1363,7 @@ static void tipc_crypto_key_synch(struct tipc_crypto *rx, struct sk_buff *skb)
  	struct tipc_msg *hdr = buf_msg(skb);
  	u32 self = tipc_own_addr(rx->net);
  	u8 cur, new;
+	unsigned long delay;
  
  	/* Update RX 'key_master' flag according to peer, also mark "legacy" if
  	 * a peer has no master key.
@@ -1322,9 +1377,22 @@ static void tipc_crypto_key_synch(struct tipc_crypto *rx, struct sk_buff *skb)
  		return;
  
  	/* Case 1: Peer has no keys, let's make master key taking over */
-	if (ehdr->rx_nokey)
+	if (ehdr->rx_nokey) {
  		/* Set or extend grace period */
  		tx->timer2 = jiffies;
+		/* Schedule key distributing for the peer if not yet */
+		if (tx->key.keys &&
+		    !atomic_cmpxchg(&rx->key_distr, 0, KEY_DISTR_SCHED)) {
+			get_random_bytes(&delay, 2);
+			delay %= 5;
+			delay = msecs_to_jiffies(500 * ++delay);
+			if (queue_delayed_work(tx->wq, &rx->work, delay))
+				tipc_node_get(rx->node);
+		}
+	} else {
+		/* Cancel a pending key distributing if any */
+		atomic_xchg(&rx->key_distr, 0);
+	}
  
  	/* Case 2: Peer RX active key has changed, let's update own TX users */
  	cur = atomic_read(&rx->peer_rx_active);
@@ -1377,6 +1445,15 @@ int tipc_crypto_start(struct tipc_crypto **crypto, struct net *net,
  	if (!c)
  		return -ENOMEM;
  
+	/* Allocate workqueue on TX */
+	if (!node) {
+		c->wq = alloc_ordered_workqueue("tipc_crypto", 0);
+		if (!c->wq) {
+			kfree(c);
+			return -ENOMEM;
+		}
+	}
+
  	/* Allocate statistic structure */
  	c->stats = alloc_percpu_gfp(struct tipc_crypto_stats, GFP_ATOMIC);
  	if (!c->stats) {
@@ -1387,7 +1464,9 @@ int tipc_crypto_start(struct tipc_crypto **crypto, struct net *net,
  	c->flags = 0;
  	c->net = net;
  	c->node = node;
+	get_random_bytes(&c->key_gen, 2);
  	tipc_crypto_key_set_state(c, 0, 0, 0);
+	atomic_set(&c->key_distr, 0);
  	atomic_set(&c->peer_rx_active, 0);
  	atomic64_set(&c->sndnxt, 0);
  	c->timer1 = jiffies;
@@ -1397,32 +1476,27 @@ int tipc_crypto_start(struct tipc_crypto **crypto, struct net *net,
  		  (is_rx(c)) ? tipc_node_get_id_str(c->node) :
  			       tipc_own_id_string(c->net));
  
+        if (is_rx(c))
+                INIT_DELAYED_WORK(&c->work, tipc_crypto_work_rx);
+
  	*crypto = c;
  	return 0;
  }
  
  void tipc_crypto_stop(struct tipc_crypto **crypto)
  {
-	struct tipc_crypto *c = *crypto, *tx, *rx;
+	struct tipc_crypto *c = *crypto;
  	u8 k;
  
  	if (!c)
  		return;
  
-	rcu_read_lock();
-	/* RX stopping? => decrease TX key users if any */
-	if (is_rx(c)) {
-		rx = c;
-		tx = tipc_net(rx->net)->crypto_tx;
-		k = atomic_read(&rx->peer_rx_active);
-		if (k) {
-			tipc_aead_users_dec(tx->aead[k], 0);
-			/* Mark the point TX key users changed */
-			tx->timer1 = jiffies;
-		}
-	}
+	/* Flush any queued works & destroy wq */
+	if (is_tx(c))
+		destroy_workqueue(c->wq);
  
  	/* Release AEAD keys */
+	rcu_read_lock();
  	for (k = KEY_MIN; k <= KEY_MAX; k++)
  		tipc_aead_put(rcu_dereference(c->aead[k]));
  	rcu_read_unlock();
@@ -1621,6 +1695,7 @@ int tipc_crypto_xmit(struct net *net, struct sk_buff **skb,
  		}
  		if (user == LINK_CONFIG ||
  		    (user == LINK_PROTOCOL && type == RESET_MSG) ||
+		    (user == MSG_CRYPTO && type == KEY_DISTR_MSG) ||
  		    time_before(jiffies, tx->timer2 + TIPC_TX_GRACE_PERIOD)) {
  			if (__rx && __rx->key_master &&
  			    !atomic_read(&__rx->peer_rx_active))
@@ -1705,7 +1780,7 @@ int tipc_crypto_rcv(struct net *net, struct tipc_crypto *rx,
  	struct tipc_aead *aead = NULL;
  	struct tipc_key key;
  	int rc = -ENOKEY;
-	u8 tx_key;
+	u8 tx_key, n;
  
  	tx_key = ((struct tipc_ehdr *)(*skb)->data)->tx_key;
  
@@ -1755,8 +1830,19 @@ int tipc_crypto_rcv(struct net *net, struct tipc_crypto *rx,
  		if (rc == -ENOKEY) {
  			kfree_skb(*skb);
  			*skb = NULL;
-			if (rx)
+			if (rx) {
+				/* Mark rx->nokey only if we dont have a
+				 * pending received session key, nor a newer
+				 * one i.e. in the next slot.
+				 */
+				n = key_next(tx_key);
+				rx->nokey = !(rx->skey ||
+					      rcu_access_pointer(rx->aead[n]));
+				pr_debug_ratelimited("%s: nokey %d, key %d/%x\n",
+						     rx->name, rx->nokey,
+						     tx_key, rx->key.keys);
  				tipc_node_put(rx->node);
+			}
  			this_cpu_inc(stats->stat[STAT_NOKEYS]);
  			return rc;
  		} else if (rc == -EBADMSG) {
@@ -2025,3 +2111,238 @@ static char *tipc_key_change_dump(struct tipc_key old, struct tipc_key new,
  	i += scnprintf(buf + i, 32 - i, "]");
  	return buf;
  }
+
+/**
+ * tipc_crypto_msg_rcv - Common 'MSG_CRYPTO' processing point
+ * @net: the struct net
+ * @skb: the receving message buffer
+ */
+void tipc_crypto_msg_rcv(struct net *net, struct sk_buff *skb)
+{
+	struct tipc_msg *hdr = buf_msg(skb);
+	struct tipc_crypto *rx;
+
+	rx = tipc_node_crypto_rx_by_addr(net, msg_prevnode(hdr));
+	if (unlikely(!rx))
+		goto exit;
+
+	switch (msg_type(hdr)) {
+	case KEY_DISTR_MSG:
+		if (tipc_crypto_key_rcv(rx, hdr))
+			goto exit;
+		break;
+	default:
+		break;
+	}
+
+	tipc_node_put(rx->node);
+
+exit:
+	kfree_skb(skb);
+}
+
+/**
+ * tipc_crypto_key_distr - Distribute a TX key
+ * @tx: the TX crypto
+ * @key: the key's index
+ * @dest: the destination tipc node, = NULL if distributing to all nodes
+ *
+ * Return: 0 in case of success, otherwise < 0
+ */
+int tipc_crypto_key_distr(struct tipc_crypto *tx, u8 key,
+			  struct tipc_node *dest)
+{
+	struct tipc_aead *aead;
+	char *dstr = (dest) ? tipc_node_get_id_str(dest) : "all";
+	u32 dnode = tipc_node_get_addr(dest);
+	int rc = -ENOKEY;
+
+	if (!sysctl_tipc_key_exchange_enabled)
+		return 0;
+
+	if (key) {
+		rcu_read_lock();
+		aead = tipc_aead_get(tx->aead[key]);
+		if (likely(aead)) {
+			rc = tipc_crypto_key_xmit(tx->net, aead->key,
+						  aead->gen, aead->mode,
+						  dnode);
+			tipc_aead_put(aead);
+		}
+		rcu_read_unlock();
+	}
+
+	if (likely(!rc))
+		pr_info("%s: key[%d] is distributed to %s\n",
+			tx->name, key, dstr);
+	else
+		pr_warn("%s: unable to distr key[%d] to %s, err %d\n",
+			tx->name, key, dstr, rc);
+
+	return rc;
+}
+
+/**
+ * tipc_crypto_key_xmit - Send a session key
+ * @net: the struct net
+ * @skey: the session key to be sent
+ * @gen: the key's generation
+ * @mode: the key's mode
+ * @dnode: the destination node address, = 0 if broadcasting to all nodes
+ *
+ * The session key 'skey' is packed in a TIPC v2 'MSG_CRYPTO/KEY_DISTR_MSG'
+ * as its data section, then xmit-ed through the uc/bc link.
+ *
+ * Return: 0 in case of success, otherwise < 0
+ */
+static int tipc_crypto_key_xmit(struct net *net, struct tipc_aead_key *skey,
+				u16 gen, u8 mode, u32 dnode)
+{
+	struct sk_buff_head pkts;
+	struct tipc_msg *hdr;
+	struct sk_buff *skb;
+	u16 size, cong_link_cnt;
+	u8 *data;
+	int rc;
+
+	size = tipc_aead_key_size(skey);
+	skb = tipc_buf_acquire(INT_H_SIZE + size, GFP_ATOMIC);
+	if (!skb)
+		return -ENOMEM;
+
+	hdr = buf_msg(skb);
+	tipc_msg_init(tipc_own_addr(net), hdr, MSG_CRYPTO, KEY_DISTR_MSG,
+		      INT_H_SIZE, dnode);
+	msg_set_size(hdr, INT_H_SIZE + size);
+	msg_set_key_gen(hdr, gen);
+	msg_set_key_mode(hdr, mode);
+
+	data = msg_data(hdr);
+	*((__be32 *)(data + TIPC_AEAD_ALG_NAME)) = htonl(skey->keylen);
+	memcpy(data, skey->alg_name, TIPC_AEAD_ALG_NAME);
+	memcpy(data + TIPC_AEAD_ALG_NAME + sizeof(__be32), skey->key,
+	       skey->keylen);
+
+	__skb_queue_head_init(&pkts);
+	__skb_queue_tail(&pkts, skb);
+	if (dnode)
+		rc = tipc_node_xmit(net, &pkts, dnode, 0);
+	else
+		rc = tipc_bcast_xmit(net, &pkts, &cong_link_cnt);
+
+	return rc;
+}
+
+/**
+ * tipc_crypto_key_rcv - Receive a session key
+ * @rx: the RX crypto
+ * @hdr: the TIPC v2 message incl. the receiving session key in its data
+ *
+ * This function retrieves the session key in the message from peer, then
+ * schedules a RX work to attach the key to the corresponding RX crypto.
+ *
+ * Return: "true" if the key has been scheduled for attaching, otherwise
+ * "false".
+ */
+static bool tipc_crypto_key_rcv(struct tipc_crypto *rx, struct tipc_msg *hdr)
+{
+	struct tipc_crypto *tx = tipc_net(rx->net)->crypto_tx;
+	struct tipc_aead_key *skey = NULL;
+	u16 key_gen = msg_key_gen(hdr);
+	u16 size = msg_data_sz(hdr);
+	u8 *data = msg_data(hdr);
+
+	spin_lock(&rx->lock);
+	if (unlikely(rx->skey || (key_gen == rx->key_gen && rx->key.keys))) {
+		pr_err("%s: key existed <%px>, gen %d vs %d\n", rx->name,
+		       rx->skey, key_gen, rx->key_gen);
+		goto exit;
+	}
+
+	/* Allocate memory for the key */
+	skey = kmalloc(size, GFP_ATOMIC);
+	if (unlikely(!skey)) {
+		pr_err("%s: unable to allocate memory for skey\n", rx->name);
+		goto exit;
+	}
+
+	/* Copy key from msg data */
+	skey->keylen = ntohl(*((__be32*)(data + TIPC_AEAD_ALG_NAME)));
+	memcpy(skey->alg_name, data, TIPC_AEAD_ALG_NAME);
+	memcpy(skey->key, data + TIPC_AEAD_ALG_NAME + sizeof(__be32),
+	       skey->keylen);
+
+	/* Sanity check */
+	if (unlikely(size != tipc_aead_key_size(skey))) {
+		kfree(skey);
+		skey = NULL;
+		goto exit;
+	}
+
+	rx->key_gen = key_gen;
+	rx->skey_mode = msg_key_mode(hdr);
+	rx->skey = skey;
+	rx->nokey = 0;
+	mb();
+
+exit:
+	spin_unlock(&rx->lock);
+
+	/* Schedule the key attaching on this crypto */
+	if (likely(skey && queue_delayed_work(tx->wq, &rx->work, 0)))
+		return true;
+
+	return false;
+}
+
+/**
+ * tipc_crypto_work_rx - Scheduled RX works handler
+ * @work: the struct RX work
+ *
+ * The function processes the previous scheduled works i.e. distributing TX key
+ * or attaching a received session key on RX crypto.
+ */
+static void tipc_crypto_work_rx(struct work_struct *work)
+{
+	struct delayed_work *dwork = to_delayed_work(work);
+	struct tipc_crypto *rx = container_of(dwork, struct tipc_crypto, work);
+	struct tipc_crypto *tx = tipc_net(rx->net)->crypto_tx;
+	unsigned long delay = msecs_to_jiffies(5000);
+	bool resched = false;
+	int rc;
+
+	/* Case 1: Distribute TX key to peer if scheduled */
+	if (atomic_cmpxchg(&rx->key_distr,
+			   KEY_DISTR_SCHED,
+			   KEY_DISTR_COMPL) == KEY_DISTR_SCHED) {
+		/* Always pick the newest one for distributing */
+		tipc_crypto_key_distr(tx,
+				      tx->key.pending ?: tx->key.active,
+				      rx->node);
+		/* Sched for key_distr releasing */
+		resched = true;
+	} else {
+		atomic_cmpxchg(&rx->key_distr, KEY_DISTR_COMPL, 0);
+	}
+
+	/* Case 2: Attach a pending received session key from peer if any */
+	if (rx->skey) {
+		rc = tipc_crypto_key_init(rx, rx->skey, rx->skey_mode, false);
+		switch (rc) {
+		case -EBUSY:
+		case -ENOMEM:
+			/* Resched the key attaching */
+			resched = true;
+			break;
+		default:
+			synchronize_rcu();
+			kfree(rx->skey);
+			rx->skey = NULL;
+		}
+	}
+
+	if (resched && queue_delayed_work(tx->wq, &rx->work, delay))
+		return;
+
+	tipc_node_put(rx->node);
+}
diff --git a/net/tipc/crypto.h b/net/tipc/crypto.h
index 7fcb80cb0e8a..70bda3d7e174 100644
--- a/net/tipc/crypto.h
+++ b/net/tipc/crypto.h
@@ -67,6 +67,7 @@ enum {
  };
  
  extern int sysctl_tipc_max_tfms __read_mostly;
+extern int sysctl_tipc_key_exchange_enabled __read_mostly;
  
  /**
   * TIPC encryption message format:
@@ -167,8 +168,31 @@ int tipc_crypto_rcv(struct net *net, struct tipc_crypto *rx,
  int tipc_crypto_key_init(struct tipc_crypto *c, struct tipc_aead_key *ukey,
  			 u8 mode, bool master_key);
  void tipc_crypto_key_flush(struct tipc_crypto *c);
+int tipc_crypto_key_distr(struct tipc_crypto *tx, u8 key,
+			  struct tipc_node *dest);
+void tipc_crypto_msg_rcv(struct net *net, struct sk_buff *skb);
  int tipc_aead_key_validate(struct tipc_aead_key *ukey);
  bool tipc_ehdr_validate(struct sk_buff *skb);
  
+static inline u32 msg_key_gen(struct tipc_msg *m)
+{
+	return msg_bits(m, 4, 16, 0xffff);
+}
+
+static inline void msg_set_key_gen(struct tipc_msg *m, u32 gen)
+{
+	msg_set_bits(m, 4, 16, 0xffff, gen);
+}
+
+static inline u32 msg_key_mode(struct tipc_msg *m)
+{
+	return msg_bits(m, 4, 0, 0xf);
+}
+
+static inline void msg_set_key_mode(struct tipc_msg *m, u32 mode)
+{
+	msg_set_bits(m, 4, 0, 0xf, mode);
+}
+
  #endif /* _TIPC_CRYPTO_H */
  #endif
diff --git a/net/tipc/link.c b/net/tipc/link.c
index 1c579357ccdf..9142e264f020 100644
--- a/net/tipc/link.c
+++ b/net/tipc/link.c
@@ -1244,6 +1244,11 @@ static bool tipc_data_input(struct tipc_link *l, struct sk_buff *skb,
  	case MSG_FRAGMENTER:
  	case BCAST_PROTOCOL:
  		return false;
+#ifdef CONFIG_TIPC_CRYPTO
+	case MSG_CRYPTO:
+		tipc_crypto_msg_rcv(l->net, skb);
+		return true;
+#endif
  	default:
  		pr_warn("Dropping received illegal msg type\n");
  		kfree_skb(skb);
diff --git a/net/tipc/msg.h b/net/tipc/msg.h
index 25e5c5c8a6ff..2b514c600c2d 100644
--- a/net/tipc/msg.h
+++ b/net/tipc/msg.h
@@ -82,6 +82,7 @@ struct plist;
  #define  NAME_DISTRIBUTOR     11
  #define  MSG_FRAGMENTER       12
  #define  LINK_CONFIG          13
+#define  MSG_CRYPTO           14
  #define  SOCK_WAKEUP          14       /* pseudo user */
  #define  TOP_SRV              15       /* pseudo user */
  
@@ -750,6 +751,11 @@ static inline void msg_set_nameupper(struct tipc_msg *m, u32 n)
  #define GRP_REMIT_MSG        5
  
  /*
+ * Crypto message types
+ */
+#define KEY_DISTR_MSG		0
+
+/*
   * Word 1
   */
  static inline u32 msg_seq_gap(struct tipc_msg *m)
diff --git a/net/tipc/node.c b/net/tipc/node.c
index 55f012d1ea74..186951b690d2 100644
--- a/net/tipc/node.c
+++ b/net/tipc/node.c
@@ -278,6 +278,14 @@ struct tipc_crypto *tipc_node_crypto_rx_by_list(struct list_head *pos)
  {
  	return container_of(pos, struct tipc_node, list)->crypto_rx;
  }
+
+struct tipc_crypto *tipc_node_crypto_rx_by_addr(struct net *net, u32 addr)
+{
+	struct tipc_node *n;
+
+	n = tipc_node_find(net, addr);
+	return (n) ? n->crypto_rx : NULL;
+}
  #endif
  
  static void tipc_node_free(struct rcu_head *rp)
@@ -303,7 +311,7 @@ void tipc_node_put(struct tipc_node *node)
  	kref_put(&node->kref, tipc_node_kref_release);
  }
  
-static void tipc_node_get(struct tipc_node *node)
+void tipc_node_get(struct tipc_node *node)
  {
  	kref_get(&node->kref);
  }
@@ -584,6 +592,7 @@ static void tipc_node_calculate_timer(struct tipc_node *n, struct tipc_link *l)
  
  static void tipc_node_delete_from_list(struct tipc_node *node)
  {
+	tipc_crypto_key_flush(node->crypto_rx);
  	list_del_rcu(&node->list);
  	hlist_del_rcu(&node->hash);
  	tipc_node_put(node);
@@ -2922,6 +2931,14 @@ static int __tipc_nl_node_set_key(struct sk_buff *skb, struct genl_info *info)
  
  	/* Initiate the TX/RX key */
  	rc = tipc_crypto_key_init(c, ukey, mode, master_key);
+	if (rc < 0 || c != tx)
+		goto exit;
+
+	/* Distribute TX key but not master one */
+	if (!master_key)
+		tipc_crypto_key_distr(tx, rc, NULL);
+
+exit:
  	if (n)
  		tipc_node_put(n);
  
diff --git a/net/tipc/node.h b/net/tipc/node.h
index 9f6f13f1604f..154a5bbb0d29 100644
--- a/net/tipc/node.h
+++ b/net/tipc/node.h
@@ -79,12 +79,14 @@ bool tipc_node_get_id(struct net *net, u32 addr, u8 *id);
  u32 tipc_node_get_addr(struct tipc_node *node);
  char *tipc_node_get_id_str(struct tipc_node *node);
  void tipc_node_put(struct tipc_node *node);
+void tipc_node_get(struct tipc_node *node);
  struct tipc_node *tipc_node_create(struct net *net, u32 addr, u8 *peer_id,
  				   u16 capabilities, u32 hash_mixes,
  				   bool preliminary);
  #ifdef CONFIG_TIPC_CRYPTO
  struct tipc_crypto *tipc_node_crypto_rx(struct tipc_node *__n);
  struct tipc_crypto *tipc_node_crypto_rx_by_list(struct list_head *pos);
+struct tipc_crypto *tipc_node_crypto_rx_by_addr(struct net *net, u32 addr);
  #endif
  u32 tipc_node_try_addr(struct net *net, u8 *id, u32 addr);
  void tipc_node_check_dest(struct net *net, u32 onode, u8 *peer_id128,
diff --git a/net/tipc/sysctl.c b/net/tipc/sysctl.c
index 97a6264a2993..9fb65c988f7f 100644
--- a/net/tipc/sysctl.c
+++ b/net/tipc/sysctl.c
@@ -74,6 +74,15 @@ static struct ctl_table tipc_table[] = {
  		.proc_handler	= proc_dointvec_minmax,
  		.extra1         = SYSCTL_ONE,
  	},
+	{
+		.procname	= "key_exchange_enabled",
+		.data		= &sysctl_tipc_key_exchange_enabled,
+		.maxlen		= sizeof(sysctl_tipc_key_exchange_enabled),
+		.mode		= 0644,
+		.proc_handler	= proc_dointvec_minmax,
+		.extra1         = SYSCTL_ZERO,
+		.extra2         = SYSCTL_ONE,
+	},
  #endif
  	{
  		.procname	= "bc_retruni",

With some clarifications as per comments above:

Acked-by: Jon Maloy <jm...@re...>

Re: [tipc-discussion] [PATCH RFC 3/5] tipc: introduce encryption master key

[Jon M. <jm...@re...>]

On 7/10/20 6:11 AM, Tuong Lien wrote:
> In addition to the supported cluster & per-node encryption keys for the
> en/decryption of TIPC messages, we now introduce one option for user to
> set a cluster key as 'master key',
Still not quite happy with the term 'master key' but have no really good 
suggestions at the moment.
Maybe 'membership key' or just 'member key'?
> which is simply a symmetric key like
> the former
but has a longer life cycle. It has two purposes:

  - Authentication of new member nodes in the cluster.
   New nodes, having no knowledge of current session keys
   in the cluster will still be able to join the cluster as long as
   they know this key. This is because all neighbor discovery
   (LINK_CONFIG) messages must be encrypted with this key.

- Encryption of session encryption keys during automatic exchange
   and update of those. This is  a feature we will introduce
   in a later commit in this series.
We insert the new key into the currently unused slot 0 in the key array,
and start using it immediately once the user has set it.
>
After joining, a node only knowing the master key should be fully
> communicable to existing nodes in the cluster, although those nodes may
> have their own session keys activated (i.e. not the master one). To
> support this, we simply
s/simply//
> define a 'grace period', starting from the time
> a node itself reports having no RX keys, so the existing nodes will use
> the master key for encryption instead. The grace period can be extended
> but will automatically stop after, e.g. 5 seconds without a new report.
> This is also the basis for later key exchanging since the new node will
> be impossible to decrypt anything without the support from master key.
>
> For user to set a master key, we define a new netlink flag -
> 'TIPC_NLA_NODE_KEY_MASTER', so it can be added to the current 'set key'
> netlink command to specify the setting key to be a master key.
>
> Above all, the traditional cluster/per-node key mechanism is guaranteed
> to work when user comes not to use this master key option. This is also
> compatible to legacy nodes without the feature supported.
Even this new key can be updated without any interruption
of cluster connectivity, but is so is needed, this has to be
coordinated and set by the user.
>
> Signed-off-by: Tuong Lien <tuo...@de...>
> ---
>   include/uapi/linux/tipc_netlink.h |   1 +
>   net/tipc/crypto.c                 | 206 ++++++++++++++++++++++++++++----------
>   net/tipc/crypto.h                 |  15 ++-
>   net/tipc/msg.h                    |   4 +-
>   net/tipc/netlink.c                |   1 +
>   net/tipc/node.c                   |  46 ++++-----
>   6 files changed, 189 insertions(+), 84 deletions(-)
>
> diff --git a/include/uapi/linux/tipc_netlink.h b/include/uapi/linux/tipc_netlink.h
> index dc0d23a50e69..d484baa9d365 100644
> --- a/include/uapi/linux/tipc_netlink.h
> +++ b/include/uapi/linux/tipc_netlink.h
> @@ -165,6 +165,7 @@ enum {
>   	TIPC_NLA_NODE_UP,		/* flag */
>   	TIPC_NLA_NODE_ID,		/* data */
>   	TIPC_NLA_NODE_KEY,		/* data */
> +	TIPC_NLA_NODE_KEY_MASTER,	/* flag */
>   
>   	__TIPC_NLA_NODE_MAX,
>   	TIPC_NLA_NODE_MAX = __TIPC_NLA_NODE_MAX - 1
> diff --git a/net/tipc/crypto.c b/net/tipc/crypto.c
> index f1046ab4bd01..7c688cd0b13e 100644
> --- a/net/tipc/crypto.c
> +++ b/net/tipc/crypto.c
> @@ -38,6 +38,7 @@
>   #include <crypto/aes.h>
>   #include "crypto.h"
>   
> +#define TIPC_TX_GRACE_PERIOD	msecs_to_jiffies(5000) /* 5s */
>   #define TIPC_TX_LASTING_TIME	msecs_to_jiffies(10000) /* 10s */
>   #define TIPC_RX_ACTIVE_LIM	msecs_to_jiffies(3000) /* 3s */
>   #define TIPC_RX_PASSIVE_LIM	msecs_to_jiffies(15000) /* 15s */
> @@ -49,9 +50,9 @@
>    * TIPC Key ids
>    */
>   enum {
> -	KEY_UNUSED = 0,
> -	KEY_MIN,
> -	KEY_1 = KEY_MIN,
> +	KEY_MASTER = 0,
> +	KEY_MIN = KEY_MASTER,
> +	KEY_1 = 1,
>   	KEY_2,
>   	KEY_3,
>   	KEY_MAX = KEY_3,
> @@ -166,27 +167,36 @@ struct tipc_crypto_stats {
>    * @aead: array of pointers to AEAD keys for encryption/decryption
>    * @peer_rx_active: replicated peer RX active key index
>    * @key: the key states
> - * @working: the crypto is working or not
>    * @stats: the crypto statistics
>    * @name: the crypto name
>    * @sndnxt: the per-peer sndnxt (TX)
>    * @timer1: general timer 1 (jiffies)
>    * @timer2: general timer 2 (jiffies)
> + * @working: the crypto is working or not
> + * @key_master: flag indicates if master key exists
> + * @legacy_user: flag indicates if a peer joins w/o master key (for bwd comp.)
>    * @lock: tipc_key lock
>    */
>   struct tipc_crypto {
>   	struct net *net;
>   	struct tipc_node *node;
> -	struct tipc_aead __rcu *aead[KEY_MAX + 1]; /* key[0] is UNUSED */
> +	struct tipc_aead __rcu *aead[KEY_MAX + 1];
>   	atomic_t peer_rx_active;
>   	struct tipc_key key;
> -	u8 working:1;
>   	struct tipc_crypto_stats __percpu *stats;
>   	char name[48];
>   
>   	atomic64_t sndnxt ____cacheline_aligned;
>   	unsigned long timer1;
>   	unsigned long timer2;
> +	union {
> +		struct {
> +			u8 working:1;
> +			u8 key_master:1;
> +			u8 legacy_user:1;
> +		};
> +		u8 flags;
> +	};
>   	spinlock_t lock; /* crypto lock */
>   
>   } ____cacheline_aligned;
> @@ -236,13 +246,19 @@ static inline void tipc_crypto_key_set_state(struct tipc_crypto *c,
>   					     u8 new_active,
>   					     u8 new_pending);
>   static int tipc_crypto_key_attach(struct tipc_crypto *c,
> -				  struct tipc_aead *aead, u8 pos);
> +				  struct tipc_aead *aead, u8 pos,
> +				  bool master_key);
>   static bool tipc_crypto_key_try_align(struct tipc_crypto *rx, u8 new_pending);
>   static struct tipc_aead *tipc_crypto_key_pick_tx(struct tipc_crypto *tx,
>   						 struct tipc_crypto *rx,
> -						 struct sk_buff *skb);
> +						 struct sk_buff *skb,
> +						 u8 tx_key);
>   static void tipc_crypto_key_synch(struct tipc_crypto *rx, struct sk_buff *skb);
>   static int tipc_crypto_key_revoke(struct net *net, u8 tx_key);
> +static inline void tipc_crypto_clone_msg(struct net *net, struct sk_buff *_skb,
> +					 struct tipc_bearer *b,
> +					 struct tipc_media_addr *dst,
> +					 struct tipc_node *__dnode, u8 type);
>   static void tipc_crypto_rcv_complete(struct net *net, struct tipc_aead *aead,
>   				     struct tipc_bearer *b,
>   				     struct sk_buff **skb, int err);
> @@ -937,8 +953,6 @@ bool tipc_ehdr_validate(struct sk_buff *skb)
>   		return false;
>   	if (unlikely(skb->len <= ehsz + TIPC_AES_GCM_TAG_SIZE))
>   		return false;
> -	if (unlikely(!ehdr->tx_key))
> -		return false;
>   
>   	return true;
>   }
> @@ -991,6 +1005,8 @@ static int tipc_ehdr_build(struct net *net, struct tipc_aead *aead,
>   	ehdr->tx_key = tx_key;
>   	ehdr->destined = (__rx) ? 1 : 0;
>   	ehdr->rx_key_active = (__rx) ? __rx->key.active : 0;
> +	ehdr->rx_nokey = (__rx) ? !__rx->key.keys : 0;
> +	ehdr->master_key = aead->crypto->key_master;
>   	ehdr->reserved_1 = 0;
>   	ehdr->reserved_2 = 0;
>   
> @@ -1033,6 +1049,7 @@ static inline void tipc_crypto_key_set_state(struct tipc_crypto *c,
>    * @c: TIPC crypto to which new key is attached
>    * @ukey: the user key
>    * @mode: the key mode (CLUSTER_KEY or PER_NODE_KEY)
> + * @master_key: specify this is a cluster master key
>    *
>    * A new TIPC AEAD key will be allocated and initiated with the specified user
>    * key, then attached to the TIPC crypto.
> @@ -1040,7 +1057,7 @@ static inline void tipc_crypto_key_set_state(struct tipc_crypto *c,
>    * Return: new key id in case of success, otherwise: < 0
>    */
>   int tipc_crypto_key_init(struct tipc_crypto *c, struct tipc_aead_key *ukey,
> -			 u8 mode)
> +			 u8 mode, bool master_key)
>   {
>   	struct tipc_aead *aead = NULL;
>   	int rc = 0;
> @@ -1053,7 +1070,7 @@ int tipc_crypto_key_init(struct tipc_crypto *c, struct tipc_aead_key *ukey,
>   	}
>   
>   	/* Attach it to the crypto */
> -	rc = tipc_crypto_key_attach(c, aead, 0);
> +	rc = tipc_crypto_key_attach(c, aead, 0, master_key);
>   	if (rc < 0) {
>   		pr_err("%s: unable to attach key, err %d\n", c->name, rc);
>   		tipc_aead_free(&aead->rcu);
> @@ -1069,11 +1086,13 @@ int tipc_crypto_key_init(struct tipc_crypto *c, struct tipc_aead_key *ukey,
>    * @c: TIPC crypto to which the new AEAD key is attached
>    * @aead: the new AEAD key pointer
>    * @pos: desired slot in the crypto key array, = 0 if any!
> + * @master_key: specify this is a cluster master key
>    *
>    * Return: new key id in case of success, otherwise: -EBUSY
>    */
>   static int tipc_crypto_key_attach(struct tipc_crypto *c,
> -				  struct tipc_aead *aead, u8 pos)
> +				  struct tipc_aead *aead, u8 pos,
> +				  bool master_key)
>   {
>   	struct tipc_key key;
>   	int rc = -EBUSY;
> @@ -1081,6 +1100,10 @@ static int tipc_crypto_key_attach(struct tipc_crypto *c,
>   
>   	spin_lock_bh(&c->lock);
>   	key = c->key;
> +	if (master_key) {
> +		new_key = KEY_MASTER;
> +		goto attach;
> +	}
>   	if (key.active && key.passive)
>   		goto exit;
>   	if (key.pending) {
> @@ -1112,8 +1135,7 @@ static int tipc_crypto_key_attach(struct tipc_crypto *c,
>   		tipc_crypto_key_set_state(c, key.passive, key.active,
>   					  key.pending);
>   	c->working = 1;
> -	c->timer1 = jiffies;
> -	c->timer2 = jiffies;
> +	c->key_master |= master_key;
>   	rc = new_key;
>   
>   exit:
> @@ -1126,7 +1148,7 @@ void tipc_crypto_key_flush(struct tipc_crypto *c)
>   	int k;
>   
>   	spin_lock_bh(&c->lock);
> -	c->working = 0;
> +	c->flags = 0;
>   	tipc_crypto_key_set_state(c, 0, 0, 0);
>   	for (k = KEY_MIN; k <= KEY_MAX; k++)
>   		tipc_crypto_key_detach(c->aead[k], &c->lock);
> @@ -1202,6 +1224,7 @@ static bool tipc_crypto_key_try_align(struct tipc_crypto *rx, u8 new_pending)
>    * @tx: TX crypto handle
>    * @rx: RX crypto handle (can be NULL)
>    * @skb: the message skb which will be decrypted later
> + * @tx_key: peer TX key id
>    *
>    * This function looks up the existing TX keys and pick one which is suitable
>    * for the message decryption, that must be a cluster key and not used before
> @@ -1211,7 +1234,8 @@ static bool tipc_crypto_key_try_align(struct tipc_crypto *rx, u8 new_pending)
>    */
>   static struct tipc_aead *tipc_crypto_key_pick_tx(struct tipc_crypto *tx,
>   						 struct tipc_crypto *rx,
> -						 struct sk_buff *skb)
> +						 struct sk_buff *skb,
> +						 u8 tx_key)
>   {
>   	struct tipc_skb_cb *skb_cb = TIPC_SKB_CB(skb);
>   	struct tipc_aead *aead = NULL;
> @@ -1230,6 +1254,10 @@ static struct tipc_aead *tipc_crypto_key_pick_tx(struct tipc_crypto *tx,
>   
>   	/* Pick one TX key */
>   	spin_lock(&tx->lock);
> +	if (tx_key == KEY_MASTER) {
> +		aead = tipc_aead_rcu_ptr(tx->aead[KEY_MASTER], &tx->lock);
> +		goto done;
> +	}
>   	do {
>   		k = (i == 0) ? key.pending :
>   			((i == 1) ? key.active : key.passive);
> @@ -1249,9 +1277,12 @@ static struct tipc_aead *tipc_crypto_key_pick_tx(struct tipc_crypto *tx,
>   		skb->next = skb_clone(skb, GFP_ATOMIC);
>   		if (unlikely(!skb->next))
>   			pr_warn("Failed to clone skb for next round if any\n");
> -		WARN_ON(!refcount_inc_not_zero(&aead->refcnt));
>   		break;
>   	} while (++i < 3);
> +
> +done:
> +	if (likely(aead))
> +		WARN_ON(!refcount_inc_not_zero(&aead->refcnt));
>   	spin_unlock(&tx->lock);
>   
>   	return aead;
> @@ -1266,6 +1297,9 @@ static struct tipc_aead *tipc_crypto_key_pick_tx(struct tipc_crypto *tx,
>    * has changed, so the number of TX keys' users on this node are increased and
>    * decreased correspondingly.
>    *
> + * It also considers if peer has no key, then we need to make own master key
> + * (if any) taking over i.e. starting grace period.
> + *
>    * The "per-peer" sndnxt is also reset when the peer key has switched.
>    */
>   static void tipc_crypto_key_synch(struct tipc_crypto *rx, struct sk_buff *skb)
> @@ -1276,11 +1310,23 @@ static void tipc_crypto_key_synch(struct tipc_crypto *rx, struct sk_buff *skb)
>   	u32 self = tipc_own_addr(rx->net);
>   	u8 cur, new;
>   
> -	/* Ensure this message is destined to us first */
> +	/* Update RX 'key_master' flag according to peer, also mark "legacy" if
> +	 * a peer has no master key.
> +	 */
> +	rx->key_master = ehdr->master_key;
> +	if (!rx->key_master)
> +		tx->legacy_user = 1;
> +
> +	/* For later cases, apply only if message is destined to this node */
>   	if (!ehdr->destined || msg_short(hdr) || msg_destnode(hdr) != self)
>   		return;
>   
> -	/* Peer RX active key has changed, let's update own TX users */
> +	/* Case 1: Peer has no keys, let's make master key taking over */
s/taking/take/
> +	if (ehdr->rx_nokey)
> +		/* Set or extend grace period */
> +		tx->timer2 = jiffies;
> +
> +	/* Case 2: Peer RX active key has changed, let's update own TX users */
>   	cur = atomic_read(&rx->peer_rx_active);
>   	new = ehdr->rx_key_active;
>   	if (tx->key.keys &&
> @@ -1338,7 +1384,7 @@ int tipc_crypto_start(struct tipc_crypto **crypto, struct net *net,
>   		return -ENOMEM;
>   	}
>   
> -	c->working = 0;
> +	c->flags = 0;
>   	c->net = net;
>   	c->node = node;
>   	tipc_crypto_key_set_state(c, 0, 0, 0);
> @@ -1473,6 +1519,12 @@ void tipc_crypto_timeout(struct tipc_crypto *rx)
>   s5:
>   	spin_unlock(&rx->lock);
>   
> +	/* Relax it here, the flag will be set again if it really is, but only
> +	 * when we are not in grace period for safety!
> +	 */
> +	if (time_after(jiffies, tx->timer2 + TIPC_TX_GRACE_PERIOD))
> +		tx->legacy_user = 0;
> +
>   	/* Limit max_tfms & do debug commands if needed */
>   	if (likely(sysctl_tipc_max_tfms <= TIPC_MAX_TFMS_LIM))
>   		return;
> @@ -1482,6 +1534,22 @@ void tipc_crypto_timeout(struct tipc_crypto *rx)
>   	tipc_crypto_do_cmd(rx->net, cmd);
>   }
>   
> +static inline void tipc_crypto_clone_msg(struct net *net, struct sk_buff *_skb,
> +					 struct tipc_bearer *b,
> +					 struct tipc_media_addr *dst,
> +					 struct tipc_node *__dnode, u8 type)
> +{
> +	struct sk_buff *skb;
> +
> +	skb = skb_clone(_skb, GFP_ATOMIC);
> +	if (skb) {
> +		TIPC_SKB_CB(skb)->xmit_type = type;
> +		tipc_crypto_xmit(net, &skb, b, dst, __dnode);
> +		if (skb)
> +			b->media->send_msg(net, skb, b, dst);
> +	}
> +}
> +
>   /**
>    * tipc_crypto_xmit - Build & encrypt TIPC message for xmit
>    * @net: struct net
> @@ -1491,7 +1559,8 @@ void tipc_crypto_timeout(struct tipc_crypto *rx)
>    * @__dnode: destination node for reference if any
>    *
>    * First, build an encryption message header on the top of the message, then
> - * encrypt the original TIPC message by using the active or pending TX key.
> + * encrypt the original TIPC message by using the pending, master or active
> + * key with this preference order.
>    * If the encryption is successful, the encrypted skb is returned directly or
>    * via the callback.
>    * Otherwise, the skb is freed!
> @@ -1514,46 +1583,63 @@ int tipc_crypto_xmit(struct net *net, struct sk_buff **skb,
>   	struct tipc_msg *hdr = buf_msg(*skb);
>   	struct tipc_key key = tx->key;
>   	struct tipc_aead *aead = NULL;
> -	struct sk_buff *_skb;
> -	int rc = -ENOKEY;
>   	u32 user = msg_user(hdr);
> -	u8 tx_key;
> +	u32 type = msg_type(hdr);
> +	int rc = -ENOKEY;
> +	u8 tx_key = 0;
>   
>   	/* No encryption? */
>   	if (!tx->working)
>   		return 0;
>   
> -	/* Try with the pending key if available and:
> -	 * 1) This is the only choice (i.e. no active key) or;
> -	 * 2) Peer has switched to this key (unicast only) or;
> -	 * 3) It is time to do a pending key probe;
> -	 */
> +	/* Pending key if peer has active on it or probing time */
>   	if (unlikely(key.pending)) {
>   		tx_key = key.pending;
> -		if (!key.active)
> +		if (!tx->key_master && !key.active)
>   			goto encrypt;
>   		if (__rx && atomic_read(&__rx->peer_rx_active) == tx_key)
>   			goto encrypt;
> -		if (TIPC_SKB_CB(*skb)->probe) {
> +		if (TIPC_SKB_CB(*skb)->xmit_type == SKB_PROBING) {
>   			pr_debug("%s: probing for key[%d]\n", tx->name,
>   				 key.pending);
>   			goto encrypt;
>   		}
> -		if (user == LINK_CONFIG || user == LINK_PROTOCOL) {
> -			_skb = skb_clone(*skb, GFP_ATOMIC);
> -			if (_skb) {
> -				TIPC_SKB_CB(_skb)->probe = 1;
> -				tipc_crypto_xmit(net, &_skb, b, dst, __dnode);
> -				if (_skb)
> -					b->media->send_msg(net, _skb, b, dst);
> +		if (user == LINK_CONFIG || user == LINK_PROTOCOL)
> +			tipc_crypto_clone_msg(net, *skb, b, dst, __dnode,
> +					      SKB_PROBING);
> +	}
> +
> +	/* Master key if this is a *vital* message or in grace period */
> +	if (tx->key_master) {
> +		tx_key = KEY_MASTER;
> +		if (!key.active)
> +			goto encrypt;
> +		if (TIPC_SKB_CB(*skb)->xmit_type == SKB_GRACING) {
> +			pr_debug("%s: gracing for msg (%d %d)\n", tx->name,
> +				 user, type);
> +			goto encrypt;
> +		}
> +		if (user == LINK_CONFIG ||
> +		    (user == LINK_PROTOCOL && type == RESET_MSG) ||
> +		    time_before(jiffies, tx->timer2 + TIPC_TX_GRACE_PERIOD)) {
Why do RESET messages need to be encrypted with the master key?
> +			if (__rx && __rx->key_master &&
> +			    !atomic_read(&__rx->peer_rx_active))
> +				goto encrypt;
> +			if (!__rx) {
> +				if (likely(!tx->legacy_user))
> +					goto encrypt;
> +				tipc_crypto_clone_msg(net, *skb, b, dst,
> +						      __dnode, SKB_GRACING);
>   			}
>   		}
>   	}
> +
>   	/* Else, use the active key if any */
>   	if (likely(key.active)) {
>   		tx_key = key.active;
>   		goto encrypt;
>   	}
> +
>   	goto exit;
>   
>   encrypt:
> @@ -1619,15 +1705,16 @@ int tipc_crypto_rcv(struct net *net, struct tipc_crypto *rx,
>   	struct tipc_aead *aead = NULL;
>   	struct tipc_key key;
>   	int rc = -ENOKEY;
> -	u8 tx_key = 0;
> +	u8 tx_key;
> +
> +	tx_key = ((struct tipc_ehdr *)(*skb)->data)->tx_key;
>   
>   	/* New peer?
>   	 * Let's try with TX key (i.e. cluster mode) & verify the skb first!
>   	 */
> -	if (unlikely(!rx))
> +	if (unlikely(!rx || tx_key == KEY_MASTER))
>   		goto pick_tx;
>   
> -	tx_key = ((struct tipc_ehdr *)(*skb)->data)->tx_key;
>   	/* Pick RX key according to TX key if any */
>   	key = rx->key;
>   	if (tx_key == key.active || tx_key == key.pending ||
> @@ -1640,7 +1727,7 @@ int tipc_crypto_rcv(struct net *net, struct tipc_crypto *rx,
>   
>   pick_tx:
>   	/* No key suitable? Try to pick one from TX... */
> -	aead = tipc_crypto_key_pick_tx(tx, rx, *skb);
> +	aead = tipc_crypto_key_pick_tx(tx, rx, *skb, tx_key);
>   	if (aead)
>   		goto decrypt;
>   	goto exit;
> @@ -1722,9 +1809,12 @@ static void tipc_crypto_rcv_complete(struct net *net, struct tipc_aead *aead,
>   				goto free_skb;
>   		}
>   
> +		/* Ignore cloning if it was TX master key */
> +		if (ehdr->tx_key == KEY_MASTER)
> +			goto rcv;
>   		if (tipc_aead_clone(&tmp, aead) < 0)
>   			goto rcv;
> -		if (tipc_crypto_key_attach(rx, tmp, ehdr->tx_key) < 0) {
> +		if (tipc_crypto_key_attach(rx, tmp, ehdr->tx_key, false) < 0) {
>   			tipc_aead_free(&tmp->rcu);
>   			goto rcv;
>   		}
> @@ -1740,10 +1830,10 @@ static void tipc_crypto_rcv_complete(struct net *net, struct tipc_aead *aead,
>   	/* Set the RX key's user */
>   	tipc_aead_users_set(aead, 1);
>   
> -rcv:
>   	/* Mark this point, RX works */
>   	rx->timer1 = jiffies;
>   
> +rcv:
>   	/* Remove ehdr & auth. tag prior to tipc_rcv() */
>   	ehdr = (struct tipc_ehdr *)(*skb)->data;
>   
> @@ -1865,14 +1955,24 @@ static char *tipc_crypto_key_dump(struct tipc_crypto *c, char *buf)
>   	char *s;
>   
>   	for (k = KEY_MIN; k <= KEY_MAX; k++) {
> -		if (k == key.passive)
> -			s = "PAS";
> -		else if (k == key.active)
> -			s = "ACT";
> -		else if (k == key.pending)
> -			s = "PEN";
> -		else
> -			s = "-";
> +		if (k == KEY_MASTER) {
> +			if (is_rx(c))
> +				continue;
> +			if (time_before(jiffies,
> +					c->timer2 + TIPC_TX_GRACE_PERIOD))
> +				s = "ACT";
> +			else
> +				s = "PAS";
> +		} else {
> +			if (k == key.passive)
> +				s = "PAS";
> +			else if (k == key.active)
> +				s = "ACT";
> +			else if (k == key.pending)
> +				s = "PEN";
> +			else
> +				s = "-";
> +		}
>   		i += scnprintf(buf + i, 200 - i, "\tKey%d: %s", k, s);
>   
>   		rcu_read_lock();
> diff --git a/net/tipc/crypto.h b/net/tipc/crypto.h
> index c3de769f49e8..7fcb80cb0e8a 100644
> --- a/net/tipc/crypto.h
> +++ b/net/tipc/crypto.h
> @@ -74,7 +74,7 @@ extern int sysctl_tipc_max_tfms __read_mostly;
>    *     3 3 2 2 2 2 2 2 2 2 2 2 1 1 1 1 1 1 1 1 1 1 0 0 0 0 0 0 0 0 0 0
>    *     1 0 9 8 7 6 5 4|3 2 1 0 9 8 7 6|5 4 3 2 1 0 9 8|7 6 5 4 3 2 1 0
>    *    +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
> - * w0:|Ver=7| User  |D|TX |RX |K|                 Rsvd                |
> + * w0:|Ver=7| User  |D|TX |RX |K|M|N|             Rsvd                |
>    *    +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
>    * w1:|                             Seqno                             |
>    * w2:|                           (8 octets)                          |
> @@ -101,6 +101,9 @@ extern int sysctl_tipc_max_tfms __read_mostly;
>    *	RX	: Currently RX active key corresponding to the destination
>    *	          node's TX key (when the "D" bit is set)
>    *	K	: Keep-alive bit (for RPS, LINK_PROTOCOL/STATE_MSG only)
> + *	M       : Bit indicates if sender has master key
> + *	N	: Bit indicates if sender has no RX keys corresponding to the
> + *	          receiver's TX (when the "D" bit is set)
>    *	Rsvd	: Reserved bit, field
>    * Word1-2:
>    *	Seqno	: The 64-bit sequence number of the encrypted message, also
> @@ -117,7 +120,9 @@ struct tipc_ehdr {
>   			__u8	destined:1,
>   				user:4,
>   				version:3;
> -			__u8	reserved_1:3,
> +			__u8	reserved_1:1,
> +				rx_nokey:1,
> +				master_key:1,
>   				keepalive:1,
>   				rx_key_active:2,
>   				tx_key:2;
> @@ -128,7 +133,9 @@ struct tipc_ehdr {
>   			__u8	tx_key:2,
>   				rx_key_active:2,
>   				keepalive:1,
> -				reserved_1:3;
> +				master_key:1,
> +				rx_nokey:1,
> +				reserved_1:1;
>   #else
>   #error  "Please fix <asm/byteorder.h>"
>   #endif
> @@ -158,7 +165,7 @@ int tipc_crypto_xmit(struct net *net, struct sk_buff **skb,
>   int tipc_crypto_rcv(struct net *net, struct tipc_crypto *rx,
>   		    struct sk_buff **skb, struct tipc_bearer *b);
>   int tipc_crypto_key_init(struct tipc_crypto *c, struct tipc_aead_key *ukey,
> -			 u8 mode);
> +			 u8 mode, bool master_key);
>   void tipc_crypto_key_flush(struct tipc_crypto *c);
>   int tipc_aead_key_validate(struct tipc_aead_key *ukey);
>   bool tipc_ehdr_validate(struct sk_buff *skb);
> diff --git a/net/tipc/msg.h b/net/tipc/msg.h
> index 1016e96db5c4..25e5c5c8a6ff 100644
> --- a/net/tipc/msg.h
> +++ b/net/tipc/msg.h
> @@ -127,7 +127,9 @@ struct tipc_skb_cb {
>   #ifdef CONFIG_TIPC_CRYPTO
>   			u8 encrypted:1;
>   			u8 decrypted:1;
> -			u8 probe:1;
> +#define SKB_PROBING	1
> +#define SKB_GRACING	2
> +			u8 xmit_type:2;
>   			u8 tx_clone_deferred:1;
>   #endif
>   		};
> diff --git a/net/tipc/netlink.c b/net/tipc/netlink.c
> index c4aee6247d55..1ec00fcc26ee 100644
> --- a/net/tipc/netlink.c
> +++ b/net/tipc/netlink.c
> @@ -108,6 +108,7 @@ const struct nla_policy tipc_nl_node_policy[TIPC_NLA_NODE_MAX + 1] = {
>   					    .len = TIPC_NODEID_LEN},
>   	[TIPC_NLA_NODE_KEY]		= { .type = NLA_BINARY,
>   					    .len = TIPC_AEAD_KEY_SIZE_MAX},
> +	[TIPC_NLA_NODE_KEY_MASTER]	= { .type = NLA_FLAG },
>   };
>   
>   /* Properties valid for media, bearer and link */
> diff --git a/net/tipc/node.c b/net/tipc/node.c
> index 030a51c4d1fa..55f012d1ea74 100644
> --- a/net/tipc/node.c
> +++ b/net/tipc/node.c
> @@ -2872,11 +2872,11 @@ static int __tipc_nl_node_set_key(struct sk_buff *skb, struct genl_info *info)
>   {
>   	struct nlattr *attrs[TIPC_NLA_NODE_MAX + 1];
>   	struct net *net = sock_net(skb->sk);
> -	struct tipc_net *tn = tipc_net(net);
> +	struct tipc_crypto *tx = tipc_net(net)->crypto_tx, *c = tx;
>   	struct tipc_node *n = NULL;
>   	struct tipc_aead_key *ukey;
> -	struct tipc_crypto *c;
> -	u8 *id, *own_id;
> +	bool master_key = false;
> +	u8 *id, *own_id, mode;
>   	int rc = 0;
>   
>   	if (!info->attrs[TIPC_NLA_NODE])
> @@ -2886,51 +2886,45 @@ static int __tipc_nl_node_set_key(struct sk_buff *skb, struct genl_info *info)
>   			      info->attrs[TIPC_NLA_NODE],
>   			      tipc_nl_node_policy, info->extack);
>   	if (rc)
> -		goto exit;
> +		return rc;
>   
>   	own_id = tipc_own_id(net);
> -	if (!own_id) {
> -		rc = -EPERM;
> -		goto exit;
> -	}
> +	if (!own_id)
> +		return -EPERM;
>   
>   	rc = tipc_nl_retrieve_key(attrs, &ukey);
>   	if (rc)
> -		goto exit;
> +		return rc;
>   
>   	rc = tipc_aead_key_validate(ukey);
>   	if (rc)
> -		goto exit;
> +		return rc;
>   
>   	rc = tipc_nl_retrieve_nodeid(attrs, &id);
>   	switch (rc) {
>   	case -ENODATA:
> -		/* Cluster key mode */
> -		rc = tipc_crypto_key_init(tn->crypto_tx, ukey, CLUSTER_KEY);
> +		mode = CLUSTER_KEY;
> +		master_key = !!(attrs[TIPC_NLA_NODE_KEY_MASTER]);
>   		break;
>   	case 0:
> -		/* Per-node key mode */
> -		if (!memcmp(id, own_id, NODE_ID_LEN)) {
> -			c = tn->crypto_tx;
> -		} else {
> +		mode = PER_NODE_KEY;
> +		if (memcmp(id, own_id, NODE_ID_LEN)) {
>   			n = tipc_node_find_by_id(net, id) ?:
>   				tipc_node_create(net, 0, id, 0xffffu, 0, true);
> -			if (unlikely(!n)) {
> -				rc = -ENOMEM;
> -				break;
> -			}
> +			if (unlikely(!n))
> +				return -ENOMEM;
>   			c = n->crypto_rx;
>   		}
> -
> -		rc = tipc_crypto_key_init(c, ukey, PER_NODE_KEY);
> -		if (n)
> -			tipc_node_put(n);
>   		break;
>   	default:
> -		break;
> +		return rc;
>   	}
>   
> -exit:
> +	/* Initiate the TX/RX key */
> +	rc = tipc_crypto_key_init(c, ukey, mode, master_key);
> +	if (n)
> +		tipc_node_put(n);
> +
>   	return (rc < 0) ? rc : 0;
>   }
>   
Acked-by: Jon Maloy <jm...@re...>

Re: [tipc-discussion] [PATCH RFC 2/5] tipc: optimize key switching time and logic

[Jon M. <jm...@re...>]

On 7/10/20 6:11 AM, Tuong Lien wrote:
> We reduce the lasting time for a pending TX key to be active as well as
> for a passive RX key to be freed which generally helps speed up the key
> switching. It is not expected to be too fast but should not be too slow
> either. Also the key handling logic is simplified that a pending RX key
> will be removed automatically if it is found not working after a number
> of times; the probing for a pending TX key is now carried on a specific
> message user ('LINK_PROTOCOL' or 'LINK_CONFIG') which is more efficient
> than using a timer on broadcast messages, the timer is reserved for use
> later as needed.
>
> The kernel logs or 'pr***()' are now made as clear as possible to user.
> Some prints are added, removed or changed to the debug-level. The
> 'TIPC_CRYPTO_DEBUG' definition is removed, and the 'pr_debug()' is used
> instead which will be much helpful in runtime.
>
> Besides we also optimize the code in some other places as a preparation
> for later commits.
>
> This commit does not change the en/decryption functionalities.
>
> Signed-off-by: Tuong Lien <tuo...@de...>
> ---
>   net/tipc/crypto.c | 344 ++++++++++++++++++++++--------------------------------
>   1 file changed, 141 insertions(+), 203 deletions(-)
>
> diff --git a/net/tipc/crypto.c b/net/tipc/crypto.c
> index 1827ce4fac5d..f1046ab4bd01 100644
> --- a/net/tipc/crypto.c
> +++ b/net/tipc/crypto.c
> @@ -38,10 +38,10 @@
>   #include <crypto/aes.h>
>   #include "crypto.h"
>   
> -#define TIPC_TX_PROBE_LIM	msecs_to_jiffies(1000) /* > 1s */
> -#define TIPC_TX_LASTING_LIM	msecs_to_jiffies(120000) /* 2 mins */
> +#define TIPC_TX_LASTING_TIME	msecs_to_jiffies(10000) /* 10s */
>   #define TIPC_RX_ACTIVE_LIM	msecs_to_jiffies(3000) /* 3s */
> -#define TIPC_RX_PASSIVE_LIM	msecs_to_jiffies(180000) /* 3 mins */
> +#define TIPC_RX_PASSIVE_LIM	msecs_to_jiffies(15000) /* 15s */
> +
>   #define TIPC_MAX_TFMS_DEF	10
>   #define TIPC_MAX_TFMS_LIM	1000
>   
> @@ -144,7 +144,7 @@ struct tipc_aead {
>   	u32 salt;
>   	u8 authsize;
>   	u8 mode;
> -	char hint[TIPC_AEAD_HINT_LEN + 1];
> +	char hint[2 * TIPC_AEAD_HINT_LEN + 1];
>   	struct rcu_head rcu;
>   
>   	atomic64_t seqno ____cacheline_aligned;
> @@ -168,9 +168,10 @@ struct tipc_crypto_stats {
>    * @key: the key states
>    * @working: the crypto is working or not
>    * @stats: the crypto statistics
> + * @name: the crypto name
>    * @sndnxt: the per-peer sndnxt (TX)
>    * @timer1: general timer 1 (jiffies)
> - * @timer2: general timer 1 (jiffies)
> + * @timer2: general timer 2 (jiffies)
>    * @lock: tipc_key lock
>    */
>   struct tipc_crypto {
> @@ -181,6 +182,7 @@ struct tipc_crypto {
>   	struct tipc_key key;
>   	u8 working:1;
>   	struct tipc_crypto_stats __percpu *stats;
> +	char name[48];
>   
>   	atomic64_t sndnxt ____cacheline_aligned;
>   	unsigned long timer1;
> @@ -239,18 +241,17 @@ static bool tipc_crypto_key_try_align(struct tipc_crypto *rx, u8 new_pending);
>   static struct tipc_aead *tipc_crypto_key_pick_tx(struct tipc_crypto *tx,
>   						 struct tipc_crypto *rx,
>   						 struct sk_buff *skb);
> -static void tipc_crypto_key_synch(struct tipc_crypto *rx, u8 new_rx_active,
> -				  struct tipc_msg *hdr);
> +static void tipc_crypto_key_synch(struct tipc_crypto *rx, struct sk_buff *skb);
>   static int tipc_crypto_key_revoke(struct net *net, u8 tx_key);
>   static void tipc_crypto_rcv_complete(struct net *net, struct tipc_aead *aead,
>   				     struct tipc_bearer *b,
>   				     struct sk_buff **skb, int err);
>   static void tipc_crypto_do_cmd(struct net *net, int cmd);
>   static char *tipc_crypto_key_dump(struct tipc_crypto *c, char *buf);
> -#ifdef TIPC_CRYPTO_DEBUG
>   static char *tipc_key_change_dump(struct tipc_key old, struct tipc_key new,
>   				  char *buf);
> -#endif
> +#define is_tx(crypto) (!(crypto)->node)
> +#define is_rx(crypto) (!is_tx(crypto))
>   
>   #define key_next(cur) ((cur) % KEY_MAX + 1)
>   
> @@ -290,7 +291,7 @@ int tipc_aead_key_validate(struct tipc_aead_key *ukey)
>   	if (unlikely(keylen != TIPC_AES_GCM_KEY_SIZE_128 &&
>   		     keylen != TIPC_AES_GCM_KEY_SIZE_192 &&
>   		     keylen != TIPC_AES_GCM_KEY_SIZE_256))
> -		return -EINVAL;
> +		return -EKEYREJECTED;
>   
>   	return 0;
>   }
> @@ -501,9 +502,9 @@ static int tipc_aead_init(struct tipc_aead **aead, struct tipc_aead_key *ukey,
>   		return err;
>   	}
>   
> -	/* Copy some chars from the user key as a hint */
> -	memcpy(tmp->hint, ukey->key, TIPC_AEAD_HINT_LEN);
> -	tmp->hint[TIPC_AEAD_HINT_LEN] = '\0';
> +	/* Form a hex string of some last bytes as the key's hint */
> +	bin2hex(tmp->hint, ukey->key + keylen - TIPC_AEAD_HINT_LEN,
> +		TIPC_AEAD_HINT_LEN);
>   
>   	/* Initialize the other data */
>   	tmp->mode = mode;
> @@ -663,13 +664,11 @@ static int tipc_aead_encrypt(struct tipc_aead *aead, struct sk_buff *skb,
>   	 * but there is no frag_list, it should be still fine!
>   	 * Otherwise, we must cow it to be a writable buffer with the tailroom.
>   	 */
> -#ifdef TIPC_CRYPTO_DEBUG
>   	SKB_LINEAR_ASSERT(skb);
>   	if (tailen > skb_tailroom(skb)) {
> -		pr_warn("TX: skb tailroom is not enough: %d, requires: %d\n",
> -			skb_tailroom(skb), tailen);
> +		pr_debug("TX(): skb tailroom is not enough: %d, requires: %d\n",
> +			 skb_tailroom(skb), tailen);
>   	}
> -#endif
>   
>   	if (unlikely(!skb_cloned(skb) && tailen <= skb_tailroom(skb))) {
>   		nsg = 1;
> @@ -1017,23 +1016,16 @@ static inline void tipc_crypto_key_set_state(struct tipc_crypto *c,
>   					     u8 new_active,
>   					     u8 new_pending)
>   {
> -#ifdef TIPC_CRYPTO_DEBUG
>   	struct tipc_key old = c->key;
>   	char buf[32];
> -#endif
>   
>   	c->key.keys = ((new_passive & KEY_MASK) << (KEY_BITS * 2)) |
>   		      ((new_active  & KEY_MASK) << (KEY_BITS)) |
>   		      ((new_pending & KEY_MASK));
>   
> -#ifdef TIPC_CRYPTO_DEBUG
> -	pr_info("%s(%s): key changing %s ::%pS\n",
> -		(c->node) ? "RX" : "TX",
> -		(c->node) ? tipc_node_get_id_str(c->node) :
> -			    tipc_own_id_string(c->net),
> -		tipc_key_change_dump(old, c->key, buf),
> -		__builtin_return_address(0));
> -#endif
> +	pr_debug("%s: key changing %s ::%pS\n", c->name,
> +		 tipc_key_change_dump(old, c->key, buf),
> +		 __builtin_return_address(0));
>   }
>   
>   /**
> @@ -1055,20 +1047,20 @@ int tipc_crypto_key_init(struct tipc_crypto *c, struct tipc_aead_key *ukey,
>   
>   	/* Initiate with the new user key */
>   	rc = tipc_aead_init(&aead, ukey, mode);
> +	if (unlikely(rc)) {
> +		pr_err("%s: unable to init key, err %d\n", c->name, rc);
> +		return rc;
> +	}
>   
>   	/* Attach it to the crypto */
> -	if (likely(!rc)) {
> -		rc = tipc_crypto_key_attach(c, aead, 0);
> -		if (rc < 0)
> -			tipc_aead_free(&aead->rcu);
> +	rc = tipc_crypto_key_attach(c, aead, 0);
> +	if (rc < 0) {
> +		pr_err("%s: unable to attach key, err %d\n", c->name, rc);
> +		tipc_aead_free(&aead->rcu);
> +		return rc;
>   	}
>   
> -	pr_info("%s(%s): key initiating, rc %d!\n",
> -		(c->node) ? "RX" : "TX",
> -		(c->node) ? tipc_node_get_id_str(c->node) :
> -			    tipc_own_id_string(c->net),
> -		rc);
> -
> +	pr_info("%s: key[%d] is successfully attached\n", c->name, rc);
>   	return rc;
>   }
>   
> @@ -1083,49 +1075,42 @@ int tipc_crypto_key_init(struct tipc_crypto *c, struct tipc_aead_key *ukey,
>   static int tipc_crypto_key_attach(struct tipc_crypto *c,
>   				  struct tipc_aead *aead, u8 pos)
>   {
> -	u8 new_pending, new_passive, new_key;
>   	struct tipc_key key;
>   	int rc = -EBUSY;
> +	u8 new_key;
>   
>   	spin_lock_bh(&c->lock);
>   	key = c->key;
>   	if (key.active && key.passive)
>   		goto exit;
> -	if (key.passive && !tipc_aead_users(c->aead[key.passive]))
> -		goto exit;
>   	if (key.pending) {
> -		if (pos)
> -			goto exit;
>   		if (tipc_aead_users(c->aead[key.pending]) > 0)
>   			goto exit;
> +		/* if (pos): ok with replacing, will be aligned when needed */
>   		/* Replace it */
> -		new_pending = key.pending;
> -		new_passive = key.passive;
> -		new_key = new_pending;
> +		new_key = key.pending;
>   	} else {
>   		if (pos) {
>   			if (key.active && pos != key_next(key.active)) {
> -				new_pending = key.pending;
> -				new_passive = pos;
> -				new_key = new_passive;
> +				key.passive = pos;
> +				new_key = pos;
>   				goto attach;
>   			} else if (!key.active && !key.passive) {
> -				new_pending = pos;
> -				new_passive = key.passive;
> -				new_key = new_pending;
> +				key.pending = pos;
> +				new_key = pos;
>   				goto attach;
>   			}
>   		}
> -		new_pending = key_next(key.active ?: key.passive);
> -		new_passive = key.passive;
> -		new_key = new_pending;
> +		key.pending = key_next(key.active ?: key.passive);
> +		new_key = key.pending;
>   	}
>   
>   attach:
>   	aead->crypto = c;
> -	tipc_crypto_key_set_state(c, new_passive, key.active, new_pending);
>   	tipc_aead_rcu_replace(c->aead[new_key], aead, &c->lock);
> -
> +	if (likely(c->key.keys != key.keys))
> +		tipc_crypto_key_set_state(c, key.passive, key.active,
> +					  key.pending);
>   	c->working = 1;
>   	c->timer1 = jiffies;
>   	c->timer2 = jiffies;
> @@ -1204,7 +1189,8 @@ static bool tipc_crypto_key_try_align(struct tipc_crypto *rx, u8 new_pending)
>   		rcu_assign_pointer(rx->aead[new_passive], tmp2);
>   	refcount_set(&tmp1->refcnt, 1);
>   	aligned = true;
> -	pr_info("RX(%s): key is aligned!\n", tipc_node_get_id_str(rx->node));
> +	pr_info_ratelimited("%s: key[%d] -> key[%d]\n", rx->name, key.pending,
> +			    new_pending);
>   
>   exit:
>   	spin_unlock(&rx->lock);
> @@ -1274,8 +1260,7 @@ static struct tipc_aead *tipc_crypto_key_pick_tx(struct tipc_crypto *tx,
>   /**
>    * tipc_crypto_key_synch: Synch own key data according to peer key status
>    * @rx: RX crypto handle
> - * @new_rx_active: latest RX active key from peer
> - * @hdr: TIPCv2 message
> + * @skb: TIPCv2 message buffer (incl. the ehdr from peer)
>    *
>    * This function updates the peer node related data as the peer RX active key
>    * has changed, so the number of TX keys' users on this node are increased and
> @@ -1283,44 +1268,35 @@ static struct tipc_aead *tipc_crypto_key_pick_tx(struct tipc_crypto *tx,
>    *
>    * The "per-peer" sndnxt is also reset when the peer key has switched.
>    */
> -static void tipc_crypto_key_synch(struct tipc_crypto *rx, u8 new_rx_active,
> -				  struct tipc_msg *hdr)
> +static void tipc_crypto_key_synch(struct tipc_crypto *rx, struct sk_buff *skb)
>   {
> -	struct net *net = rx->net;
> -	struct tipc_crypto *tx = tipc_net(net)->crypto_tx;
> -	u8 cur_rx_active;
> -
> -	/* TX might be even not ready yet */
> -	if (unlikely(!tx->key.active && !tx->key.pending))
> -		return;
> -
> -	cur_rx_active = atomic_read(&rx->peer_rx_active);
> -	if (likely(cur_rx_active == new_rx_active))
> -		return;
> +	struct tipc_ehdr *ehdr = (struct tipc_ehdr *)skb_network_header(skb);
> +	struct tipc_crypto *tx = tipc_net(rx->net)->crypto_tx;
> +	struct tipc_msg *hdr = buf_msg(skb);
> +	u32 self = tipc_own_addr(rx->net);
> +	u8 cur, new;
>   
> -	/* Make sure this message destined for this node */
> -	if (unlikely(msg_short(hdr) ||
> -		     msg_destnode(hdr) != tipc_own_addr(net)))
> +	/* Ensure this message is destined to us first */
> +	if (!ehdr->destined || msg_short(hdr) || msg_destnode(hdr) != self)
>   		return;
>   
> -	/* Peer RX active key has changed, try to update owns' & TX users */
> -	if (atomic_cmpxchg(&rx->peer_rx_active,
> -			   cur_rx_active,
> -			   new_rx_active) == cur_rx_active) {
> -		if (new_rx_active)
> -			tipc_aead_users_inc(tx->aead[new_rx_active], INT_MAX);
> -		if (cur_rx_active)
> -			tipc_aead_users_dec(tx->aead[cur_rx_active], 0);
> +	/* Peer RX active key has changed, let's update own TX users */
> +	cur = atomic_read(&rx->peer_rx_active);
> +	new = ehdr->rx_key_active;
> +	if (tx->key.keys &&
> +	    cur != new &&
> +	    atomic_cmpxchg(&rx->peer_rx_active, cur, new) == cur) {
> +		if (new)
> +			tipc_aead_users_inc(tx->aead[new], INT_MAX);
> +		if (cur)
> +			tipc_aead_users_dec(tx->aead[cur], 0);
>   
>   		atomic64_set(&rx->sndnxt, 0);
>   		/* Mark the point TX key users changed */
>   		tx->timer1 = jiffies;
>   
> -#ifdef TIPC_CRYPTO_DEBUG
> -		pr_info("TX(%s): key users changed %d-- %d++, peer RX(%s)\n",
> -			tipc_own_id_string(net), cur_rx_active,
> -			new_rx_active, tipc_node_get_id_str(rx->node));
> -#endif
> +		pr_debug("%s: key users changed %d-- %d++, peer %s\n",
> +			 tx->name, cur, new, rx->name);
>   	}
>   }
>   
> @@ -1338,7 +1314,7 @@ static int tipc_crypto_key_revoke(struct net *net, u8 tx_key)
>   	tipc_crypto_key_detach(tx->aead[key.active], &tx->lock);
>   	spin_unlock(&tx->lock);
>   
> -	pr_warn("TX(%s): key is revoked!\n", tipc_own_id_string(net));
> +	pr_warn("%s: key is revoked\n", tx->name);
>   	return -EKEYREVOKED;
>   }
>   
> @@ -1371,25 +1347,26 @@ int tipc_crypto_start(struct tipc_crypto **crypto, struct net *net,
>   	c->timer1 = jiffies;
>   	c->timer2 = jiffies;
>   	spin_lock_init(&c->lock);
> -	*crypto = c;
> +	scnprintf(c->name, 48, "%s(%s)", (is_rx(c)) ? "RX" : "TX",
> +		  (is_rx(c)) ? tipc_node_get_id_str(c->node) :
> +			       tipc_own_id_string(c->net));
>   
> +	*crypto = c;
>   	return 0;
>   }
>   
>   void tipc_crypto_stop(struct tipc_crypto **crypto)
>   {
> -	struct tipc_crypto *c, *tx, *rx;
> -	bool is_rx;
> +	struct tipc_crypto *c = *crypto, *tx, *rx;
>   	u8 k;
>   
> -	if (!*crypto)
> +	if (!c)
>   		return;
>   
>   	rcu_read_lock();
>   	/* RX stopping? => decrease TX key users if any */
> -	is_rx = !!((*crypto)->node);
> -	if (is_rx) {
> -		rx = *crypto;
> +	if (is_rx(c)) {
> +		rx = c;
>   		tx = tipc_net(rx->net)->crypto_tx;
>   		k = atomic_read(&rx->peer_rx_active);
>   		if (k) {
> @@ -1400,15 +1377,10 @@ void tipc_crypto_stop(struct tipc_crypto **crypto)
>   	}
>   
>   	/* Release AEAD keys */
> -	c = *crypto;
>   	for (k = KEY_MIN; k <= KEY_MAX; k++)
>   		tipc_aead_put(rcu_dereference(c->aead[k]));
>   	rcu_read_unlock();
> -
> -	pr_warn("%s(%s) has been purged, node left!\n",
> -		(is_rx) ? "RX" : "TX",
> -		(is_rx) ? tipc_node_get_id_str((*crypto)->node) :
> -			  tipc_own_id_string((*crypto)->net));
> +	pr_debug("%s: has been stopped\n", c->name);
>   
>   	/* Free this crypto statistics */
>   	free_percpu(c->stats);
> @@ -1422,102 +1394,81 @@ void tipc_crypto_timeout(struct tipc_crypto *rx)
>   	struct tipc_net *tn = tipc_net(rx->net);
>   	struct tipc_crypto *tx = tn->crypto_tx;
>   	struct tipc_key key;
> -	u8 new_pending, new_passive;
>   	int cmd;
>   
> -	/* TX key activating:
> -	 * The pending key (users > 0) -> active
> -	 * The active key if any (users == 0) -> free
> -	 */
> +	/* TX pending: taking all users & stable -> active */
>   	spin_lock(&tx->lock);
>   	key = tx->key;
>   	if (key.active && tipc_aead_users(tx->aead[key.active]) > 0)
>   		goto s1;
>   	if (!key.pending || tipc_aead_users(tx->aead[key.pending]) <= 0)
>   		goto s1;
> -	if (time_before(jiffies, tx->timer1 + TIPC_TX_LASTING_LIM))
> +	if (time_before(jiffies, tx->timer1 + TIPC_TX_LASTING_TIME))
>   		goto s1;
>   
>   	tipc_crypto_key_set_state(tx, key.passive, key.pending, 0);
>   	if (key.active)
>   		tipc_crypto_key_detach(tx->aead[key.active], &tx->lock);
>   	this_cpu_inc(tx->stats->stat[STAT_SWITCHES]);
> -	pr_info("TX(%s): key %d is activated!\n", tipc_own_id_string(tx->net),
> -		key.pending);
> +	pr_info("%s: key[%d] is activated\n", tx->name, key.pending);
>   
>   s1:
>   	spin_unlock(&tx->lock);
>   
> -	/* RX key activating:
> -	 * The pending key (users > 0) -> active
> -	 * The active key if any -> passive, freed later
> -	 */
> +	/* RX pending: having user -> active */
>   	spin_lock(&rx->lock);
>   	key = rx->key;
>   	if (!key.pending || tipc_aead_users(rx->aead[key.pending]) <= 0)
>   		goto s2;
>   
> -	new_pending = (key.passive &&
> -		       !tipc_aead_users(rx->aead[key.passive])) ?
> -				       key.passive : 0;
> -	new_passive = (key.active) ?: ((new_pending) ? 0 : key.passive);
> -	tipc_crypto_key_set_state(rx, new_passive, key.pending, new_pending);
> +	if (key.active)
> +		key.passive = key.active;
> +	key.active = key.pending;
> +	rx->timer2 = jiffies;
> +	tipc_crypto_key_set_state(rx, key.passive, key.active, 0);
>   	this_cpu_inc(rx->stats->stat[STAT_SWITCHES]);
> -	pr_info("RX(%s): key %d is activated!\n",
> -		tipc_node_get_id_str(rx->node),	key.pending);
> +	pr_info("%s: key[%d] is activated\n", rx->name, key.pending);
>   	goto s5;
>   
>   s2:
> -	/* RX key "faulty" switching:
> -	 * The faulty pending key (users < -30) -> passive
> -	 * The passive key (users = 0) -> pending
> -	 * Note: This only happens after RX deactivated - s3!
> -	 */
> -	key = rx->key;
> -	if (!key.pending || tipc_aead_users(rx->aead[key.pending]) > -30)
> -		goto s3;
> -	if (!key.passive || tipc_aead_users(rx->aead[key.passive]) != 0)
> +	/* RX pending: not working -> remove */
> +	if (!key.pending || tipc_aead_users(rx->aead[key.pending]) > -10)
>   		goto s3;
>   
> -	new_pending = key.passive;
> -	new_passive = key.pending;
> -	tipc_crypto_key_set_state(rx, new_passive, key.active, new_pending);
> +	tipc_crypto_key_set_state(rx, key.passive, key.active, 0);
> +	tipc_crypto_key_detach(rx->aead[key.pending], &rx->lock);
> +	pr_info("%s: key[%d] is removed\n", rx->name, key.pending);
>   	goto s5;
>   
>   s3:
> -	/* RX key deactivating:
> -	 * The passive key if any -> pending
> -	 * The active key -> passive (users = 0) / pending
> -	 * The pending key if any -> passive (users = 0)
> -	 */
> -	key = rx->key;
> +	/* RX active: timed out or no user -> pending */
>   	if (!key.active)
>   		goto s4;
> -	if (time_before(jiffies, rx->timer1 + TIPC_RX_ACTIVE_LIM))
> +	if (time_before(jiffies, rx->timer1 + TIPC_RX_ACTIVE_LIM) &&
> +	    tipc_aead_users(rx->aead[key.active]) > 0)
>   		goto s4;
>   
> -	new_pending = (key.passive) ?: key.active;
> -	new_passive = (key.passive) ? key.active : key.pending;
> -	tipc_aead_users_set(rx->aead[new_pending], 0);
> -	if (new_passive)
> -		tipc_aead_users_set(rx->aead[new_passive], 0);
> -	tipc_crypto_key_set_state(rx, new_passive, 0, new_pending);
> -	pr_info("RX(%s): key %d is deactivated!\n",
> -		tipc_node_get_id_str(rx->node), key.active);
> +	if (key.pending)
> +		key.passive = key.active;
> +	else
> +		key.pending = key.active;
> +	rx->timer2 = jiffies;
> +	tipc_crypto_key_set_state(rx, key.passive, 0, key.pending);
> +	tipc_aead_users_set(rx->aead[key.pending], 0);
> +	pr_info("%s: key[%d] is deactivated\n", rx->name, key.active);
>   	goto s5;
>   
>   s4:
> -	/* RX key passive -> freed: */
> -	key = rx->key;
> -	if (!key.passive || !tipc_aead_users(rx->aead[key.passive]))
> +	/* RX passive: outdated or not working -> free */
> +	if (!key.passive)
>   		goto s5;
> -	if (time_before(jiffies, rx->timer2 + TIPC_RX_PASSIVE_LIM))
> +	if (time_before(jiffies, rx->timer2 + TIPC_RX_PASSIVE_LIM) &&
> +	    tipc_aead_users(rx->aead[key.passive]) > -10)
>   		goto s5;
>   
>   	tipc_crypto_key_set_state(rx, 0, key.active, key.pending);
>   	tipc_crypto_key_detach(rx->aead[key.passive], &rx->lock);
> -	pr_info("RX(%s): key %d is freed!\n", tipc_node_get_id_str(rx->node),
> -		key.passive);
> +	pr_info("%s: key[%d] is freed\n", rx->name, key.passive);
>   
>   s5:
>   	spin_unlock(&rx->lock);
> @@ -1560,10 +1511,12 @@ int tipc_crypto_xmit(struct net *net, struct sk_buff **skb,
>   	struct tipc_crypto *__rx = tipc_node_crypto_rx(__dnode);
>   	struct tipc_crypto *tx = tipc_net(net)->crypto_tx;
>   	struct tipc_crypto_stats __percpu *stats = tx->stats;
> +	struct tipc_msg *hdr = buf_msg(*skb);
>   	struct tipc_key key = tx->key;
>   	struct tipc_aead *aead = NULL;
> -	struct sk_buff *probe;
> +	struct sk_buff *_skb;
>   	int rc = -ENOKEY;
> +	u32 user = msg_user(hdr);
>   	u8 tx_key;
>   
>   	/* No encryption? */
> @@ -1581,17 +1534,18 @@ int tipc_crypto_xmit(struct net *net, struct sk_buff **skb,
>   			goto encrypt;
>   		if (__rx && atomic_read(&__rx->peer_rx_active) == tx_key)
>   			goto encrypt;
> -		if (TIPC_SKB_CB(*skb)->probe)
> +		if (TIPC_SKB_CB(*skb)->probe) {
> +			pr_debug("%s: probing for key[%d]\n", tx->name,
> +				 key.pending);
>   			goto encrypt;
> -		if (!__rx &&
> -		    time_after(jiffies, tx->timer2 + TIPC_TX_PROBE_LIM)) {
> -			tx->timer2 = jiffies;
> -			probe = skb_clone(*skb, GFP_ATOMIC);
> -			if (probe) {
> -				TIPC_SKB_CB(probe)->probe = 1;
> -				tipc_crypto_xmit(net, &probe, b, dst, __dnode);
> -				if (probe)
> -					b->media->send_msg(net, probe, b, dst);
> +		}
> +		if (user == LINK_CONFIG || user == LINK_PROTOCOL) {
> +			_skb = skb_clone(*skb, GFP_ATOMIC);
> +			if (_skb) {
> +				TIPC_SKB_CB(_skb)->probe = 1;
> +				tipc_crypto_xmit(net, &_skb, b, dst, __dnode);
> +				if (_skb)
> +					b->media->send_msg(net, _skb, b, dst);
>   			}
>   		}
>   	}
> @@ -1673,22 +1627,12 @@ int tipc_crypto_rcv(struct net *net, struct tipc_crypto *rx,
>   	if (unlikely(!rx))
>   		goto pick_tx;
>   
> -	/* Pick RX key according to TX key, three cases are possible:
> -	 * 1) The current active key (likely) or;
> -	 * 2) The pending (new or deactivated) key (if any) or;
> -	 * 3) The passive or old active key (i.e. users > 0);
> -	 */
>   	tx_key = ((struct tipc_ehdr *)(*skb)->data)->tx_key;
> +	/* Pick RX key according to TX key if any */
>   	key = rx->key;
> -	if (likely(tx_key == key.active))
> +	if (tx_key == key.active || tx_key == key.pending ||
> +	    tx_key == key.passive)
>   		goto decrypt;
> -	if (tx_key == key.pending)
> -		goto decrypt;
> -	if (tx_key == key.passive) {
> -		rx->timer2 = jiffies;
> -		if (tipc_aead_users(rx->aead[key.passive]) > 0)
> -			goto decrypt;
> -	}
>   
>   	/* Unknown key, let's try to align RX key(s) */
>   	if (tipc_crypto_key_try_align(rx, tx_key))
> @@ -1747,21 +1691,17 @@ static void tipc_crypto_rcv_complete(struct net *net, struct tipc_aead *aead,
>   	struct tipc_aead *tmp = NULL;
>   	struct tipc_ehdr *ehdr;
>   	struct tipc_node *n;
> -	u8 rx_key_active;
> -	bool destined;
>   
>   	/* Is this completed by TX? */
> -	if (unlikely(!rx->node)) {
> +	if (unlikely(is_tx(aead->crypto))) {
>   		rx = skb_cb->tx_clone_ctx.rx;
> -#ifdef TIPC_CRYPTO_DEBUG
> -		pr_info("TX->RX(%s): err %d, aead %p, skb->next %p, flags %x\n",
> -			(rx) ? tipc_node_get_id_str(rx->node) : "-", err, aead,
> -			(*skb)->next, skb_cb->flags);
> -		pr_info("skb_cb [recurs %d, last %p], tx->aead [%p %p %p]\n",
> -			skb_cb->tx_clone_ctx.recurs, skb_cb->tx_clone_ctx.last,
> -			aead->crypto->aead[1], aead->crypto->aead[2],
> -			aead->crypto->aead[3]);
> -#endif
> +		pr_debug("TX->RX(%s): err %d, aead %p, skb->next %p, flags %x\n",
> +			 (rx) ? tipc_node_get_id_str(rx->node) : "-", err, aead,
> +			 (*skb)->next, skb_cb->flags);
> +		pr_debug("skb_cb [recurs %d, last %p], tx->aead [%p %p %p]\n",
> +			 skb_cb->tx_clone_ctx.recurs, skb_cb->tx_clone_ctx.last,
> +			 aead->crypto->aead[1], aead->crypto->aead[2],
> +			 aead->crypto->aead[3]);
>   		if (unlikely(err)) {
>   			if (err == -EBADMSG && (*skb)->next)
>   				tipc_rcv(net, (*skb)->next, b);
> @@ -1782,9 +1722,6 @@ static void tipc_crypto_rcv_complete(struct net *net, struct tipc_aead *aead,
>   				goto free_skb;
>   		}
>   
> -		/* Skip cloning this time as we had a RX pending key */
> -		if (rx->key.pending)
> -			goto rcv;
>   		if (tipc_aead_clone(&tmp, aead) < 0)
>   			goto rcv;
>   		if (tipc_crypto_key_attach(rx, tmp, ehdr->tx_key) < 0) {
> @@ -1809,8 +1746,12 @@ static void tipc_crypto_rcv_complete(struct net *net, struct tipc_aead *aead,
>   
>   	/* Remove ehdr & auth. tag prior to tipc_rcv() */
>   	ehdr = (struct tipc_ehdr *)(*skb)->data;
> -	destined = ehdr->destined;
> -	rx_key_active = ehdr->rx_key_active;
> +
> +	/* Mark this point, RX passive still works */
> +	if (rx->key.passive && ehdr->tx_key == rx->key.passive)
> +		rx->timer2 = jiffies;
> +
> +	skb_reset_network_header(*skb);
>   	skb_pull(*skb, tipc_ehdr_size(ehdr));
>   	pskb_trim(*skb, (*skb)->len - aead->authsize);
>   
> @@ -1820,9 +1761,8 @@ static void tipc_crypto_rcv_complete(struct net *net, struct tipc_aead *aead,
>   		goto free_skb;
>   	}
>   
> -	/* Update peer RX active key & TX users */
> -	if (destined)
> -		tipc_crypto_key_synch(rx, rx_key_active, buf_msg(*skb));
> +	/* Ok, everything's fine, try to synch own keys according to peers' */
> +	tipc_crypto_key_synch(rx, *skb);
>   
>   	/* Mark skb decrypted */
>   	skb_cb->decrypted = 1;
> @@ -1881,7 +1821,7 @@ static void tipc_crypto_do_cmd(struct net *net, int cmd)
>   	/* Print crypto statistics */
>   	for (i = 0, j = 0; i < MAX_STATS; i++)
>   		j += scnprintf(buf + j, 200 - j, "|%11s ", hstats[i]);
> -	pr_info("\nCounter     %s", buf);
> +	pr_info("Counter     %s", buf);
>   
>   	memset(buf, '-', 115);
>   	buf[115] = '\0';
> @@ -1939,7 +1879,7 @@ static char *tipc_crypto_key_dump(struct tipc_crypto *c, char *buf)
>   		aead = rcu_dereference(c->aead[k]);
>   		if (aead)
>   			i += scnprintf(buf + i, 200 - i,
> -				       "{\"%s...\", \"%s\"}/%d:%d",
> +				       "{\"0x...%s\", \"%s\"}/%d:%d",
>   				       aead->hint,
>   				       (aead->mode == CLUSTER_KEY) ? "c" : "p",
>   				       atomic_read(&aead->users),
> @@ -1948,14 +1888,13 @@ static char *tipc_crypto_key_dump(struct tipc_crypto *c, char *buf)
>   		i += scnprintf(buf + i, 200 - i, "\n");
>   	}
>   
> -	if (c->node)
> +	if (is_rx(c))
>   		i += scnprintf(buf + i, 200 - i, "\tPeer RX active: %d\n",
>   			       atomic_read(&c->peer_rx_active));
>   
>   	return buf;
>   }
>   
> -#ifdef TIPC_CRYPTO_DEBUG
>   static char *tipc_key_change_dump(struct tipc_key old, struct tipc_key new,
>   				  char *buf)
>   {
> @@ -1986,4 +1925,3 @@ static char *tipc_key_change_dump(struct tipc_key old, struct tipc_key new,
>   	i += scnprintf(buf + i, 32 - i, "]");
>   	return buf;
>   }
> -#endif

Acked-by: Jon Maloy <jm...@re...>

Happy you also got rid of some of the #ifdef TIPC_CRYPTO_DEBUG macros

#ifdef TIPC_CRYPTO_DEBUG

Re: [tipc-discussion] [PATCH RFC 1/5] tipc: fix using smp_processor_id() in preemptible

[Jon M. <jm...@re...>]

On 7/10/20 6:11 AM, Tuong Lien wrote:
> The 'this_cpu_ptr()' is used to obtain the AEAD key' TFM on the current
> CPU for encryption, however the execution can be preemptible since it's
> actually user-space context, so the 'using smp_processor_id() in
> preemptible' has been observed.
>
> We fix the issue by using the 'get/put_cpu_ptr()' API which consists of
> a 'preempt_disable()' instead.
>
> Signed-off-by: Tuong Lien <tuo...@de...>
> ---
>   net/tipc/crypto.c | 12 +++++++++---
>   1 file changed, 9 insertions(+), 3 deletions(-)
>
> diff --git a/net/tipc/crypto.c b/net/tipc/crypto.c
> index c8c47fc72653..1827ce4fac5d 100644
> --- a/net/tipc/crypto.c
> +++ b/net/tipc/crypto.c
> @@ -326,7 +326,8 @@ static void tipc_aead_free(struct rcu_head *rp)
>   	if (aead->cloned) {
>   		tipc_aead_put(aead->cloned);
>   	} else {
> -		head = *this_cpu_ptr(aead->tfm_entry);
> +		head = *get_cpu_ptr(aead->tfm_entry);
> +		put_cpu_ptr(aead->tfm_entry);
>   		list_for_each_entry_safe(tfm_entry, tmp, &head->list, list) {
>   			crypto_free_aead(tfm_entry->tfm);
>   			list_del(&tfm_entry->list);
> @@ -399,10 +400,15 @@ static void tipc_aead_users_set(struct tipc_aead __rcu *aead, int val)
>    */
>   static struct crypto_aead *tipc_aead_tfm_next(struct tipc_aead *aead)
>   {
> -	struct tipc_tfm **tfm_entry = this_cpu_ptr(aead->tfm_entry);
> +	struct tipc_tfm **tfm_entry;
> +	struct crypto_aead *tfm;
>   
> +	tfm_entry = get_cpu_ptr(aead->tfm_entry);
>   	*tfm_entry = list_next_entry(*tfm_entry, list);
> -	return (*tfm_entry)->tfm;
> +	tfm = (*tfm_entry)->tfm;
> +	put_cpu_ptr(tfm_entry);
> +
> +	return tfm;
>   }
>   
>   /**
Acked-by: Jon Maloy <jm...@re...>

I think you should send this as a separate patch to 'net'.
Don't forget to add a "Fixes: ..." line to the commit log.

///jon

[tipc-discussion] Fwd: Re: TIPC/RAFT?

[Jon M. <jm...@re...>]

Adding tipc-discussion and tipc-dek to the thread.


-------- Forwarded Message --------
Subject: 	Re: TIPC/RAFT?
Date: 	Sat, 18 Jul 2020 12:07:18 -0400
From: 	Jon Maloy <jm...@re...>
To: 	Xue, Ying <Yin...@wi...>



Hi Ying,
I may be ranting now...
I know that blockchain technologies (Bitcoin, Ethereum etc.) also can be 
used to implement consensus algorithms. Furthermore, those are 
completely decentralized and very scalable, something that fits much 
better into the TIPC philosophy than the leader electing algorithms of 
Paxos and Raft.
It might be worth learning more about this before embarking on any major 
effort to re-implement Raft.
I don't know much about blockchains yet, so I have no clue if this is at 
all feasible regarding code volumes, consensus times and other 
properties, but it should at least be considered before any other track 
is selected.
I will try to read up on this on my side.

Regards
///jon


On 7/15/20 11:10 AM, Xue, Ying wrote:
> Hi Jon,
>
> Raft protocol is much more complex than my initial image. The 
> complexity mainly comes from: 1.Raft protocol itself is quite complex, 
> so it's not very easy to totally understand it in a very detailed way; 
> 2. Implementing a raft consensus algorithm might not be very 
> difficult, but making sure its implementation's quality reaches an 
> industrial criteria level is very difficult. However, for a consensus 
> protocol, it must ensure it can work very well in any kinds of 
> different conditions, otherwise, some distributed values become 
> inconsistent.
>
> When it comes to this implementation 
> (https://github.com/willemt/raft), I took a lot time to study, 
> validate and understand its code so far. Although I have make all test 
> cases contained in this implementation passed, the scenarios covered 
> by these test cases are quite basic particularly compared to test 
> cases included in etcd 
> (https://github.com/etcd-io/etcd/tree/master/raft). Even in its code, 
> it clearly comments some corner cases haven't handled yet.
>
> In my previous plan, I wanted to take shortcuts so that I can quickly 
> port the implementation into Linux kernel, but it was found it's very 
> difficult to properly adapt it into kernel space if I couldn't totally 
> understand Raft protocol itself, which wasted my lots of time.
>
> Currently I have to study the following two papers:
> https://raft.github.io/raft.pdf
> https://github.com/ongardie/dissertation/blob/master/stanford.pdf
>
> While reading and understanding the papers, I am researching the code 
> and test cases of "etcd", which can help to have a much deeper 
> understanding of the protocol itself. Honestly, I still don't have a 
> very well understanding for the protocol. When I read paper, I felt I 
> had a very good understanding for some parts of the protocol, but when 
> I came back to research its implementations, I lost again and again.
>
> With my understanding of the protocol becoming deeper and deeper, more 
> and more concerns with the quality of https://github.com/willemt/raft, 
> appear in my mind. In sum, I need more time to evaluate its quality 
> and then consider whether we can use it as a base to implement raft 
> protocol in TIPC stack or if it's possible for us to well implement 
> the protocol in kernel space within a not very long time.
>
> Although I have to admit studying and even implementing the protocol 
> is a pretty challenging thing, I am still quite interested in 
> continuing working on this topic.
>
> Thanks,
> Ying
>
> -----Original Message-----
> From: Jon Maloy <jm...@re...>
> Sent: Saturday, July 11, 2020 2:29 AM
> To: Xue, Ying <Yin...@wi...>
> Subject: TIPC/RAFT?
>
> Hi Ying,
> Since I haven't had any news from you for a while, I was just getting 
> curious.
> Did you have any time to look into the RAFT implementation references 
> I sent you?
> Do you have any thougths about this?
>
> BR
> ///jon
>

Re: [tipc-discussion] [net v1 1/1] tipc: allow to build NACK message in link timeout function

[Jon M. <jm...@re...>]

On 7/16/20 1:13 AM, Tung Nguyen wrote:
> Commit 02288248b051 ("tipc: eliminate gap indicator from ACK messages")
> eliminated sending of the 'gap' indicator in regular ACK messages and
> only allowed to build NACK message with enabled probe/probe_reply.
> However, necessary correction for building NACK message was missed
> in tipc_link_timeout() function. This leads to significant delay and
> link reset (due to retransmission failure) in lossy environment.
>
> This commit fixes it by setting the 'probe' flag to 'true' when
> the receive deferred queue is not empty. As a result, NACK message
> will be built to send back to another peer.
>
> Fixes: commit 02288248b051 ("tipc: eliminate gap indicator from ACK messages")
> Signed-off-by: Tung Nguyen <tun...@de...>
> ---
>   net/tipc/link.c | 2 +-
>   1 file changed, 1 insertion(+), 1 deletion(-)
>
> diff --git a/net/tipc/link.c b/net/tipc/link.c
> index 263d950e70e9..d40f8e5b7683 100644
> --- a/net/tipc/link.c
> +++ b/net/tipc/link.c
> @@ -827,11 +827,11 @@ int tipc_link_timeout(struct tipc_link *l, struct sk_buff_head *xmitq)
>   		state |= l->bc_rcvlink->rcv_unacked;
>   		state |= l->rcv_unacked;
>   		state |= !skb_queue_empty(&l->transmq);
> -		state |= !skb_queue_empty(&l->deferdq);
>   		probe = mstate->probing;
>   		probe |= l->silent_intv_cnt;
>   		if (probe || mstate->monitoring)
>   			l->silent_intv_cnt++;
> +		probe |= !skb_queue_empty(&l->deferdq);
>   		if (l->snd_nxt == l->checkpoint) {
>   			tipc_link_update_cwin(l, 0, 0);
>   			probe = true;
Acked-by: Jon Maloy <jm...@re...>

[tipc-discussion] [net v1 1/1] tipc: allow to build NACK message in link timeout function

[Tung N. <tun...@de...>]

Commit 02288248b051 ("tipc: eliminate gap indicator from ACK messages")
eliminated sending of the 'gap' indicator in regular ACK messages and
only allowed to build NACK message with enabled probe/probe_reply.
However, necessary correction for building NACK message was missed
in tipc_link_timeout() function. This leads to significant delay and
link reset (due to retransmission failure) in lossy environment.

This commit fixes it by setting the 'probe' flag to 'true' when
the receive deferred queue is not empty. As a result, NACK message
will be built to send back to another peer.

Fixes: commit 02288248b051 ("tipc: eliminate gap indicator from ACK messages")
Signed-off-by: Tung Nguyen <tun...@de...>
---
 net/tipc/link.c | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/net/tipc/link.c b/net/tipc/link.c
index 263d950e70e9..d40f8e5b7683 100644
--- a/net/tipc/link.c
+++ b/net/tipc/link.c
@@ -827,11 +827,11 @@ int tipc_link_timeout(struct tipc_link *l, struct sk_buff_head *xmitq)
 		state |= l->bc_rcvlink->rcv_unacked;
 		state |= l->rcv_unacked;
 		state |= !skb_queue_empty(&l->transmq);
-		state |= !skb_queue_empty(&l->deferdq);
 		probe = mstate->probing;
 		probe |= l->silent_intv_cnt;
 		if (probe || mstate->monitoring)
 			l->silent_intv_cnt++;
+		probe |= !skb_queue_empty(&l->deferdq);
 		if (l->snd_nxt == l->checkpoint) {
 			tipc_link_update_cwin(l, 0, 0);
 			probe = true;
-- 
2.17.1

Re: [tipc-discussion] [PATCH v2] tipc: Don't using smp_processor_id() in preemptible code

[Tuong T. L. <tuo...@de...>]

> -----Original Message-----
> From: Zhang, Qiang <Qia...@wi...>
> Sent: Wednesday, July 15, 2020 9:13 AM
> To: Eric Dumazet <eri...@gm...>; jm...@re...; da...@da...; ku...@ke...; Tuong Tong Lien
> <tuo...@de...>; Xue, Ying <Yin...@wi...>
> Cc: ne...@vg...; tip...@li...; lin...@vg...
> Subject: 回复: [PATCH v2] tipc: Don't using smp_processor_id() in preemptible code
> 
> 
> 
> ________________________________________
> 发件人: Eric Dumazet <eri...@gm...>
> 发送时间: 2020年7月14日 22:15
> 收件人: Zhang, Qiang; jm...@re...; da...@da...; ku...@ke...; tuo...@de...;
> eri...@gm...; Xue, Ying
> 抄送: ne...@vg...; tip...@li...; lin...@vg...
> 主题: Re: [PATCH v2] tipc: Don't using smp_processor_id() in preemptible code
> 
> 
> 
> On 7/14/20 1:05 AM, qia...@wi... wrote:
> > From: Zhang Qiang <qia...@wi...>
> >
> > CPU: 0 PID: 6801 Comm: syz-executor201 Not tainted 5.8.0-rc4-syzkaller #0
> > Hardware name: Google Google Compute Engine/Google Compute Engine,
> > BIOS Google 01/01/2011
> >
> > Fixes: fc1b6d6de2208 ("tipc: introduce TIPC encryption & authentication")
> > Reported-by: syz...@sy...
> > Signed-off-by: Zhang Qiang <qia...@wi...>
> > ---
> >  v1->v2:
> >  add fixes tags.
> >
> >  net/tipc/crypto.c | 3 ++-
> >  1 file changed, 2 insertions(+), 1 deletion(-)
> >
> > diff --git a/net/tipc/crypto.c b/net/tipc/crypto.c
> > index 8c47ded2edb6..520af0afe1b3 100644
> > --- a/net/tipc/crypto.c
> > +++ b/net/tipc/crypto.c
> > @@ -399,9 +399,10 @@ static void tipc_aead_users_set(struct tipc_aead __rcu *aead, int val)
> >   */
> >  static struct crypto_aead *tipc_aead_tfm_next(struct tipc_aead *aead)
> >  {
> > -     struct tipc_tfm **tfm_entry = this_cpu_ptr(aead->tfm_entry);
> > +     struct tipc_tfm **tfm_entry = get_cpu_ptr(aead->tfm_entry);
> >
> >       *tfm_entry = list_next_entry(*tfm_entry, list);
> > +     put_cpu_ptr(tfm_entry);
> >       return (*tfm_entry)->tfm;
> >  }
> >
> >
> 
> > You have not explained why this was safe.
> >
> >  This seems to hide a real bug.
> >
> > Presumably callers of this function should have disable preemption, and maybe > interrupts as well.
> >
> >Right after put_cpu_ptr(tfm_entry), this thread could migrate to another cpu, >and still access
> >data owned by the old cpu.
> 
> Thanks for you suggest, I will check code again.
> 

Actually, last week I sent a similar patch to tipc-discussion which covers the
case as well (there is also another place causing the same issue...). If you
don't mind, you can take a look at below (just copied/pasted).

BR/Tuong

-----Original Message-----
From: Tuong Tong Lien <tuo...@de...> 
Sent: Friday, July 10, 2020 5:11 PM
To: jm...@re...; ma...@do...; yin...@wi...; tip...@li...
Cc: tipc-dek <tip...@de...>
Subject: [PATCH RFC 1/5] tipc: fix using smp_processor_id() in preemptible

The 'this_cpu_ptr()' is used to obtain the AEAD key' TFM on the current
CPU for encryption, however the execution can be preemptible since it's
actually user-space context, so the 'using smp_processor_id() in
preemptible' has been observed.

We fix the issue by using the 'get/put_cpu_ptr()' API which consists of
a 'preempt_disable()' instead.

Signed-off-by: Tuong Lien <tuo...@de...>
---
 net/tipc/crypto.c | 12 +++++++++---
 1 file changed, 9 insertions(+), 3 deletions(-)

diff --git a/net/tipc/crypto.c b/net/tipc/crypto.c
index c8c47fc72653..1827ce4fac5d 100644
--- a/net/tipc/crypto.c
+++ b/net/tipc/crypto.c
@@ -326,7 +326,8 @@ static void tipc_aead_free(struct rcu_head *rp)
 	if (aead->cloned) {
 		tipc_aead_put(aead->cloned);
 	} else {
-		head = *this_cpu_ptr(aead->tfm_entry);
+		head = *get_cpu_ptr(aead->tfm_entry);
+		put_cpu_ptr(aead->tfm_entry);
 		list_for_each_entry_safe(tfm_entry, tmp, &head->list, list) {
 			crypto_free_aead(tfm_entry->tfm);
 			list_del(&tfm_entry->list);
@@ -399,10 +400,15 @@ static void tipc_aead_users_set(struct tipc_aead __rcu *aead, int val)
  */
 static struct crypto_aead *tipc_aead_tfm_next(struct tipc_aead *aead)
 {
-	struct tipc_tfm **tfm_entry = this_cpu_ptr(aead->tfm_entry);
+	struct tipc_tfm **tfm_entry;
+	struct crypto_aead *tfm;
 
+	tfm_entry = get_cpu_ptr(aead->tfm_entry);
 	*tfm_entry = list_next_entry(*tfm_entry, list);
-	return (*tfm_entry)->tfm;
+	tfm = (*tfm_entry)->tfm;
+	put_cpu_ptr(tfm_entry);
+
+	return tfm;
 }
 
 /**
-- 
2.13.7