add nonce to protect replay attack

PCC.DevClient/DevClientApp.cs

@@ -1,4 +1,5 @@
-using PCC.App.Security;
+using System.Security.Cryptography;
+using PCC.App.Security;
 using PCC.Common.AssemblyInject.Interfaces;
 using PCC.Common.EventBus;
 using PCC.App;
@@ -12,6 +13,8 @@ internal class DevClientApp(DevClientPccConfigManager configManager, TrustedPeer
         logger.LogInformation("init");
         eventBus.Subscript<TrustedPeerManager.TPM_EVT_PEER_CX>(OnConnected);
         eventBus.Subscript<TrustedPeerManager.TPM_EVT_PEER_RX>(OnRx);
+        eventBus.Subscript<TrustedPeerManager.TPM_EVT_PEER_DX>(OnDx);
+        eventBus.Subscript<TrustedPeerManager.TPM_EVT_PEER_XX>(OnXx);
     }
 
     public void Start()
@@ -47,19 +50,31 @@ internal class DevClientApp(DevClientPccConfigManager configManager, TrustedPeer
         var tph = configManager.Instance.TrustPeerHost!;
         var tpp = configManager.Instance.TrustPeerPort!.Value;
 
-        var peerId = tpm.AddTrustPeer(tpk, tph, tpp);
+        var peerId = tpm.AddPeer(tpk, tph, tpp);
         tpm.ConnectToPeerAsync(peerId);
     }
 
     private void OnConnected(TrustedPeerManager.TPM_EVT_PEER_CX obj)
     {
         logger.LogInformation("Connected");
-        tpm.SendToPeer(obj.PeerId, "Brr连上了？"u8.ToArray());
+        var payload = "Brr连上了？"u8.ToArray();
+        logger.LogInformation($"Send payload {Convert.ToHexString(SHA256.HashData(payload))}");
+        tpm.SendToPeer(obj.PeerId, payload);
     }
 
     private void OnRx(TrustedPeerManager.TPM_EVT_PEER_RX obj)
     {
-        logger.LogInformation("Rx:" + Convert.ToHexString(obj.block));
+        logger.LogInformation("Rx:" + Convert.ToHexString(obj.payload.Span));
+    }
+
+    private void OnDx(TrustedPeerManager.TPM_EVT_PEER_DX obj)
+    {
+        logger.LogInformation("Dx:" + obj.PeerId);
+    }
+
+    private void OnXx(TrustedPeerManager.TPM_EVT_PEER_XX obj)
+    {
+        logger.LogWarning($"有内鬼，终止交易！ {obj.Kind} {obj.PeerId}");
     }
 
     public void Stop()

PCC.DevServer/DevServerApp.cs

@@ -24,12 +24,9 @@ internal class DevServerApp(
         eventBus.Subscript<TrustedPeerManager.TPM_EVT_PEER_IX>(OnIncome);
         eventBus.Subscript<TrustedPeerManager.TPM_EVT_PEER_RX>(OnRx);
         eventBus.Subscript<TrustedPeerManager.TPM_EVT_PEER_DX>(OnDx);
-
+        eventBus.Subscript<TrustedPeerManager.TPM_EVT_PEER_XX>(OnXx);
     }
 
-
-
-
     public void Start()
     {
         logger.LogInformation("starting...");
@@ -48,7 +45,6 @@ internal class DevServerApp(
         {
             myPri = Convert.FromBase64String(configManager.Instance.MyKeyPrivate);
             myPub = Convert.FromBase64String(configManager.Instance.MyKeyPublic);
-
         }
 
         eventBus.Publish(new TrustedPeerManager.TPM_EVT_CMD_INIT(myPub, myPri));
@@ -68,7 +64,7 @@ internal class DevServerApp(
         }
 
         //warn just example for server
-        tpm.AddTrustPeer(Convert.FromBase64String(configManager.Instance.TrustPeerKeyPub), "", 0);
+        tpm.AddPeer(Convert.FromBase64String(configManager.Instance.TrustPeerKeyPub), "", 0);
 
         _tcpServer = new KestrelTcpServer(configManager.Instance.ListenAddress, configManager.Instance.ListenPort.Value, tpm.HandleIncomingPeerAsync, ktsLogger);
         Task.Run(async () =>
@@ -89,12 +85,11 @@ internal class DevServerApp(
         logger.LogInformation("Income:" + obj.PeerId);
     }
 
-
     private void OnRx(TrustedPeerManager.TPM_EVT_PEER_RX obj)
     {
         logger.LogInformation($"Rx from <{obj.PeerId}>");
-        logger.LogInformation($"Rx content: {Encoding.UTF8.GetString(obj.block)}");
-        tpm.SendToPeer(obj.PeerId, SHA256.HashData(obj.block));
+        logger.LogInformation($"Rx content: {Encoding.UTF8.GetString(obj.payload.Span)}");
+        tpm.SendToPeer(obj.PeerId, SHA256.HashData(obj.payload.Span));
     }
 
     private void OnDx(TrustedPeerManager.TPM_EVT_PEER_DX obj)
@@ -102,6 +97,11 @@ internal class DevServerApp(
         logger.LogInformation("Disconnected:" + obj.PeerId);
     }
 
+    private void OnXx(TrustedPeerManager.TPM_EVT_PEER_XX obj)
+    {
+        logger.LogWarning($"有内鬼，终止交易！ {obj.Kind} {obj.PeerId}");
+    }
+
     public void Stop()
     {
         logger.LogInformation("stop");

PCC.Shared/App/BinaryFormatter/BinaryRwExtensionMethod.cs

@@ -23,6 +23,16 @@ public static class BinaryRwExtensionMethod
         writer.Write(data.Span);
     }
 
+    public static void Write(this BinaryWriter writer, DateTimeOffset dateTimeOffset)
+    {
+        writer.Write(dateTimeOffset.ToUnixTimeMilliseconds());
+    }
+
+    public static DateTimeOffset ReadDateTimeOffset(this BinaryReader reader)
+    {
+        return DateTimeOffset.FromUnixTimeMilliseconds(reader.ReadInt64());
+    }
+
     /// <summary>
     /// 向 PipeWriter 中写入 7 位编码的整数。
     /// </summary>

PCC.Shared/App/Networking/TcpPeer.cs

@@ -1,11 +1,10 @@
 using System.Buffers;
-using System.Net.Sockets;
 using Microsoft.AspNetCore.Connections;
 using PCC.App.BinaryFormatter;
 
 namespace PCC.App.Networking;
 
-public class TcpPeer(ConnectionContext context, Socket? sck)
+public class TcpPeer(ConnectionContext context)
 {
     public async Task SendBlockAsync(ReadOnlyMemory<byte> block)
     {
@@ -65,7 +64,7 @@ public class TcpPeer(ConnectionContext context, Socket? sck)
 
     public void Disconnect()
     {
-        if (sck != null) sck.Disconnect(false);
-        else context.Abort(new ConnectionAbortedException("Disconnect"));
+        context.Transport.Input.Complete();
+        context.Transport.Output.Complete();
     }
 }

PCC.Shared/App/Security/EncryptedTcpPeer.cs

@@ -5,13 +5,13 @@ namespace PCC.App.Security;
 
 public class EncryptedTcpPeer(RSA senderPri, RSA destPub, TcpPeer peer)
 {
-    public async Task SendBlock(ReadOnlyMemory<byte> block)
+    public async Task SendBlockAsync(ReadOnlyMemory<byte> block)
     {
         var eb = RsaUtility.EncryptAndSignature(destPub, block.Span, senderPri);
         await peer.SendBlockAsync(eb);
     }
 
-    public async Task SendBlock(byte[] block)
+    public async Task SendBlockAsync(byte[] block)
     {
         var eb = RsaUtility.EncryptAndSignature(destPub, block, senderPri);
         await peer.SendBlockAsync(eb);

PCC.Shared/App/Security/TimestampNonceManager.cs

@@ -0,0 +1,120 @@
+using System.Buffers.Binary;
+using System.Collections.Concurrent;
+using System.Security.Cryptography;
+
+namespace PCC.App.Security;
+
+public class TimestampNonceManager : IDisposable
+{
+    public static int TimestampLength => sizeof(long);
+
+    private readonly ConcurrentDictionary<string, DateTime> _holds = new();
+    private readonly TimeSpan _expiration;
+    private readonly TimeSpan _maxTimeSkew;
+    private readonly Timer _cleanupTimer;
+
+    private readonly int _randomLength;
+
+    public int NonceLength { get; }
+
+    private bool _disposed;
+
+    public TimestampNonceManager(int randomLength, TimeSpan expiration, TimeSpan maxTimeSkew)
+    {
+        NonceLength = TimestampLength + randomLength;
+        _randomLength = randomLength;
+        _expiration = expiration;
+        _maxTimeSkew = maxTimeSkew;
+        _cleanupTimer = new Timer(CleanupTimerCallback, null, expiration, expiration);
+    }
+
+    /// <summary> 生成一个新的nonce </summary>
+    public (ReadOnlyMemory<byte> nonce, DateTimeOffset timestamp) NewNonce()
+    {
+        ThrowIfDisposed();
+
+        var nonce = new byte[NonceLength];
+
+        // 把时间戳放在nonce开头
+        var timestamp = DateTimeOffset.UtcNow;
+        BinaryPrimitives.TryWriteInt64LittleEndian(nonce, timestamp.ToUnixTimeMilliseconds());
+
+        // 剩下的字节作为随机 nonce
+        RandomNumberGenerator.Fill(nonce.AsSpan(TimestampLength));
+
+        return (new ReadOnlyMemory<byte>(nonce), timestamp);
+    }
+
+    /// <summary> 生成一个新的nonce并追加payload </summary>
+    public (ReadOnlyMemory<byte> payloadAndNonce, DateTimeOffset timestamp) NewNonceWithPayload(ReadOnlyMemory<byte> payload)
+    {
+        ThrowIfDisposed();
+        var payloadAndNonce = new byte[NonceLength + payload.Length];
+
+        // 把时间戳放在nonce开头
+        var timestamp = DateTimeOffset.UtcNow;
+        BinaryPrimitives.TryWriteInt64LittleEndian(payloadAndNonce, timestamp.ToUnixTimeMilliseconds());
+
+        // 剩下的字节作为随机 nonce
+        RandomNumberGenerator.Fill(payloadAndNonce.AsSpan(TimestampLength, _randomLength));
+
+        // 追加 payload
+        payload.Span.CopyTo(payloadAndNonce.AsSpan(NonceLength));
+
+        return (new ReadOnlyMemory<byte>(payloadAndNonce), timestamp);
+    }
+
+    /// <summary> 验证nonce： null时间误差过大，false检测到重放攻击，true没问题 </summary>
+    public bool? CheckValid(ReadOnlyMemory<byte> timestampNonce, out DateTimeOffset timestamp)
+    {
+        ThrowIfDisposed();
+
+        if (timestampNonce.Length != NonceLength) throw new ArgumentException("invalid length", nameof(timestampNonce));
+
+        timestamp = DateTimeOffset.FromUnixTimeMilliseconds(BinaryPrimitives.ReadInt64LittleEndian(timestampNonce.Span[..TimestampLength]));
+
+        // 检查时间戳是否在允许的时间窗口内,超过最大时间差则拒绝
+        if (Math.Abs((DateTimeOffset.UtcNow - timestamp).TotalMilliseconds) > _maxTimeSkew.TotalMilliseconds) return null;
+
+        // 剩下的部分为真正的随机 nonce
+        var nonce = Convert.ToHexString(timestampNonce[TimestampLength..].Span);
+
+        // 如果 nonce 已存在，则为重放攻击，返回 false；否则添加并返回 true
+        return _holds.TryAdd(nonce, DateTime.UtcNow);
+    }
+
+    /// <summary> 验证nonce并提取payload： null时间误差过大，false检测到重放攻击，true没问题 </summary>
+    public (bool?, DateTimeOffset timestamp, ReadOnlyMemory<byte> payload) CheckValidAndExtractPayload(ReadOnlyMemory<byte> payloadAndNonce)
+    {
+        ThrowIfDisposed();
+
+        if (payloadAndNonce.Length < NonceLength) throw new ArgumentException("invalid length", nameof(payloadAndNonce));
+
+        //提取时间戳、nonce、payload
+        var timestamp = DateTimeOffset.FromUnixTimeMilliseconds(BinaryPrimitives.ReadInt64LittleEndian(payloadAndNonce.Span[..TimestampLength]));
+        var nonce = Convert.ToHexString(payloadAndNonce.Slice(TimestampLength, _randomLength).Span);
+        var payload = payloadAndNonce[NonceLength..];
+
+        // 检查时间戳是否在允许的时间窗口内, 超过最大时间差则拒绝
+        if (Math.Abs((DateTimeOffset.UtcNow - timestamp).TotalMilliseconds) > _maxTimeSkew.TotalMilliseconds) return (null, timestamp, payload);
+
+        // 如果 nonce 已存在，则为重放攻击，返回 false；否则添加并返回 true
+        return (_holds.TryAdd(nonce, DateTime.UtcNow), timestamp, payload);
+    }
+
+    private void CleanupTimerCallback(object? state)
+    {
+        var now = DateTime.UtcNow;
+        foreach (var item in _holds.Where(p => now - p.Value >= _expiration)) _holds.TryRemove(item.Key, out _);
+    }
+
+    private void ThrowIfDisposed() => ObjectDisposedException.ThrowIf(_disposed, this);
+
+    public void Dispose()
+    {
+        if (_disposed) return;
+        _cleanupTimer.Dispose();
+        _disposed = true;
+        GC.SuppressFinalize(this);
+    }
+}

PCC.Shared/App/TransferModels/BlockWithNonce.cs

@@ -0,0 +1,32 @@
+using System.Text;
+using PCC.App.BinaryFormatter;
+
+namespace PCC.App.TransferModels;
+
+public static class BlockWithNonce1
+{
+    public static ReadOnlyMemory<byte> Mux(ReadOnlyMemory<byte> payload, ReadOnlyMemory<byte> nonce)
+    {
+        // 计算总长度
+        var totalLength = payload.Length + nonce.Length;
+        var buffer = new byte[totalLength];
+
+        // 将 payload 和 nonce 拷贝到 buffer 中
+        payload.Span.CopyTo(buffer);
+        nonce.Span.CopyTo(buffer.AsSpan(payload.Length));
+
+        return buffer;
+    }
+
+    public static (ReadOnlyMemory<byte> payload, ReadOnlyMemory<byte> nonce) Demux(ReadOnlyMemory<byte> transferBlock, int nonceBytes)
+    {
+        // 计算 payload 的长度
+        var payloadLength = transferBlock.Length - nonceBytes;
+
+        // 获取 payload 和 nonce
+        var payload = transferBlock.Slice(0, payloadLength);
+        var nonce = transferBlock.Slice(payloadLength, nonceBytes);
+
+        return (payload, nonce);
+    }
+}

PCC.Shared/App/TrustedPeerManager.cs

@@ -5,6 +5,7 @@ using Microsoft.AspNetCore.Connections;
 using Microsoft.AspNetCore.Server.Kestrel.Transport.Sockets;
 using PCC.App.Networking;
 using PCC.App.Security;
+using PCC.App.TransferModels;
 using PCC.Common.EventBus;
 
 namespace PCC.App;
@@ -20,31 +21,49 @@ namespace PCC.App;
 
 public abstract class TrustedPeerManager
 {
+    private const int NONCE_LENGTH_BYTES = 16;
+    private const int NONCE_EXPIRE_SECOND = 60;
+    private const int NONCE_SKEW_SECOND = 30;
+
     public record TPM_EVT_CMD_INIT(byte[] MyPub, byte[] MyPri);
     public record TPM_EVT_CMD_SHUTDOWN;
 
-    public record TPM_EVT_PEER_CX(string PeerId);
     public record TPM_EVT_PEER_IX(string PeerId);
+    public record TPM_EVT_PEER_CX(string PeerId);
     public record TPM_EVT_PEER_DX(string PeerId);
-    public record TPM_EVT_PEER_RX(string PeerId, byte[] block);
+    public record TPM_EVT_PEER_RX(string PeerId, DateTimeOffset senderTimestamp, ReadOnlyMemory<byte> payload);
+    public record TPM_EVT_PEER_TX(string PeerId, DateTimeOffset senderTimestamp, ReadOnlyMemory<byte> payload);
+
+    public record TPM_EVT_PEER_XX(XX Kind, string? PeerId);
+
+    public enum XX
+    {
+        Invalid = 0,
+        NonTrustPeer,
+        HandshakeFailServerAckNotMatch,
+        TimeStampSkew,
+        ReplayAttackDetected,
+    }
 
     private readonly IEventBus _eventBus;
     private readonly ILogger<TrustedPeerManager> _logger;
     private readonly SocketConnectionContextFactory _connectionContextFactory;
 
-    private RSA _myPrivateKey;
-    private ReadOnlyMemory<byte> _myPeerId;
+    private RSA? _myPrivateKey;
+    private ReadOnlyMemory<byte>? _myPeerId;
 
     private readonly ConcurrentDictionary<string, PeerInfo> _trustedPeers = new();
 
     private readonly ConcurrentDictionary<string, EncryptedTcpPeer> _connectedPeers = new();
     private readonly ConcurrentDictionary<string, EncryptedTcpPeer> _IncomePeers = new();
+    private readonly TimestampNonceManager _nonceManager;
 
     protected TrustedPeerManager(IEventBus eventBus, ILogger<TrustedPeerManager> logger)
     {
         _eventBus = eventBus;
         _logger = logger;
         _connectionContextFactory = new(new(), _logger);
+        _nonceManager = new TimestampNonceManager(NONCE_LENGTH_BYTES - TimestampNonceManager.TimestampLength, TimeSpan.FromMicroseconds(NONCE_EXPIRE_SECOND), TimeSpan.FromSeconds(NONCE_SKEW_SECOND));
 
         eventBus.Subscript<TPM_EVT_CMD_INIT>(Init);
         eventBus.Subscript<TPM_EVT_CMD_SHUTDOWN>(Shutdown);
@@ -58,25 +77,30 @@ public abstract class TrustedPeerManager
 
     private void Shutdown(TPM_EVT_CMD_SHUTDOWN _)
     {
-        foreach (var item in _connectedPeers.Values)
+        foreach (var item in _connectedPeers)
         {
-            item.Disconnect();
+            item.Value.Disconnect();
+            _connectedPeers.Remove(item.Key, out var _);
         }
 
-        foreach (var item in _IncomePeers.Values)
+        foreach (var item in _IncomePeers)
         {
-            item.Disconnect();
+            item.Value.Disconnect();
+            _connectedPeers.Remove(item.Key, out var _);
         }
+
+        _nonceManager.Dispose();
     }
 
-    public string AddTrustPeer(byte[] tpk, string host, int port)
+    public string AddPeer(byte[] publicKey, string host, int port)
     {
-        var peerId = Convert.ToHexString(SHA256.HashData(tpk));
-        var rsa = RsaUtility.FromPKCS1PublicKey(tpk);
-        _trustedPeers[peerId] = new PeerInfo(rsa, host, port);
+        var peerId = Convert.ToHexString(SHA256.HashData(publicKey));
+        _trustedPeers[peerId] = new PeerInfo(RsaUtility.FromPKCS1PublicKey(publicKey), host, port);
         return peerId;
     }
 
+    public bool RemovePeer(string peerId) => _trustedPeers.Remove(peerId, out _);
+
     public bool ConnectToPeerAsync(string peerId)
     {
         if (_trustedPeers.TryGetValue(peerId, out var peerInfo) == false)
@@ -84,49 +108,87 @@ public abstract class TrustedPeerManager
             return false;
         }
 
+        if (_myPrivateKey == null || _myPeerId == null) throw new InvalidOperationException("Not init");
+
         _ = Task.Run(async () =>
         {
+            EncryptedTcpPeer? epLocal = null;
             try
             {
-                //Start Peer client
-                // connect to server
-                using var sck = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
-                await sck.ConnectAsync(peerInfo.host, peerInfo.port);
-
-                var conn = _connectionContextFactory.Create(sck);
-                var peer = new TcpPeer(conn, sck);
-
-                // handshake as client
-
-                var ep = new EncryptedTcpPeer(_myPrivateKey, peerInfo.PublicKey, peer);
-                await ep.SendBlock(_myPeerId); //send handshake 1
-
-                var blockAck = await ep.RxBlockAsync(new CancellationTokenSource(TimeSpan.FromSeconds(10)).Token);
-                if (blockAck?.SequenceEqual(SHA256.HashData(_myPeerId.Span)) != true)
+                //Create Peer client and connect to server
                 {
-                    peer.Disconnect();
-                    return;
+                    var sckLocal = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
+                    await sckLocal.ConnectAsync(peerInfo.Host, peerInfo.Port);
+
+                    var conn = _connectionContextFactory.Create(sckLocal);
+                    var tcpPeerLocal = new TcpPeer(conn);
+                    epLocal = new EncryptedTcpPeer(_myPrivateKey, peerInfo.PublicKey, tcpPeerLocal);
+
+                    // handshake as client
+
+                    // ☞ tx handshake 1
+                    var (handshake1, _) = _nonceManager.NewNonceWithPayload(_myPeerId.Value);
+                    await epLocal.SendBlockAsync(handshake1);
+
+                    // ☞ rx handshake 2
+                    var handshake2 = await epLocal.RxBlockAsync(new CancellationTokenSource(TimeSpan.FromSeconds(10)).Token);
+                    if (handshake2 == null) { epLocal.Disconnect(); return; } // timeout? disconnect
+
+                    var (ok, _, pl) = _nonceManager.CheckValidAndExtractPayload(handshake2);
+                    if (ok != true)
+                    {
+                        PublishXX(peerId, ok switch { null => XX.TimeStampSkew, false => XX.ReplayAttackDetected, _ => XX.Invalid });
+                        epLocal.Disconnect();
+                        return;
+                    }
+
+                    // handshake not match, disconnect
+                    if (pl.Span.SequenceEqual(SHA256.HashData(handshake1.Span)) != true)
+                    {
+                        PublishXX(peerId, XX.HandshakeFailServerAckNotMatch);
+                        epLocal.Disconnect(); return;
+                    }
                 }
 
+                //replace exist connection
+                if (_connectedPeers.Remove(peerId, out var exist)) exist.Disconnect();
+
                 // register as connected
-                //TODO: replace exist connection
-                _connectedPeers[peerId] = ep;
+                _connectedPeers[peerId] = epLocal;
 
                 _eventBus.Publish(new TPM_EVT_PEER_CX(peerId));
 
                 while (true) // start RxCycle
                 {
-                    var rxb = await ep.RxBlockAsync();
+                    var rxb = await epLocal.RxBlockAsync();
                     if (rxb == null) break;
-                    _eventBus.Publish(new TPM_EVT_PEER_RX(peerId, rxb));
+
+                    var (ok, senderTimestamp, payload) = _nonceManager.CheckValidAndExtractPayload(rxb);
+                    if (ok != true)
+                    {
+                        PublishXX(peerId, ok switch { null => XX.TimeStampSkew, false => XX.ReplayAttackDetected, _ => XX.Invalid });
+                        epLocal.Disconnect();
+                        break;
+                    }
+
+                    _eventBus.Publish(new TPM_EVT_PEER_RX(peerId, senderTimestamp, payload));
                 }
             }
+            catch (ConnectionAbortedException)
+            {
+                //ignore
+            }
             catch (Exception ex)
             {
                 _logger.LogError(ex, nameof(ConnectToPeerAsync));
             }
 
             _eventBus.Publish(new TPM_EVT_PEER_DX(peerId));
+
+            // unregister from connected
+            _connectedPeers.Remove(peerId, out _);
+
+            epLocal?.Disconnect();
         });
 
         return true;
@@ -134,51 +196,111 @@ public abstract class TrustedPeerManager
 
     public async Task HandleIncomingPeerAsync(ConnectionContext connection)
     {
+        if (_myPrivateKey == null) throw new InvalidOperationException("Not init");
+
         string? peerId = null;
         try
         {
-            var peer = new TcpPeer(connection, null);
-
             // handshake as server
-            var eb = await peer.RxBlockAsync(new CancellationTokenSource(TimeSpan.FromSeconds(10)).Token);
-            if (eb == null) return;
+            EncryptedTcpPeer epRemote;
+            {
+                var tpRemote = new TcpPeer(connection);
+                //rx handshake1
+                PeerInfo peerInfo;
+                byte[] payloadOfHandshake1;
+                {
+                    var receivedBlock = await tpRemote.RxBlockAsync(new CancellationTokenSource(TimeSpan.FromSeconds(10)).Token);
+                    if (receivedBlock == null) return;
+
+                    var handshake1 = RsaUtility.Decrypt(_myPrivateKey, receivedBlock);
+
+                    var (ok, _, payload) = _nonceManager.CheckValidAndExtractPayload(handshake1);
+                    if (ok != true)
+                    {
+                        PublishXX(null, ok switch { null => XX.TimeStampSkew, false => XX.ReplayAttackDetected, _ => XX.Invalid });
+                        tpRemote.Disconnect();
+                        return;
+                    }
+
+                    peerId = Convert.ToHexString(payload.Span);
+
+                    // Kick if non-trusted
+                    if (_trustedPeers.TryGetValue(peerId, out peerInfo!) == false)
+                    {
+                        PublishXX(peerId, XX.NonTrustPeer);
+                        return;
+                    }
+
+                    // verify signature
+                    payloadOfHandshake1 = RsaUtility.DecryptAndVerifySignature(_myPrivateKey, receivedBlock, peerInfo.PublicKey);
+                }
 
-            // extract peerId
-            peerId = Convert.ToHexString(RsaUtility.Decrypt(_myPrivateKey, eb));
-            // Kick if non trusted
-            if (_trustedPeers.TryGetValue(peerId, out var peerInfo) == false) return;
-            // verify signature
-            var data = RsaUtility.DecryptAndVerifySignature(_myPrivateKey, eb, peerInfo.PublicKey);
+                //tx handshake2
+                {
+                    epRemote = new EncryptedTcpPeer(_myPrivateKey, peerInfo.PublicKey, tpRemote);
+                    var payloadOfHandshake2 = SHA256.HashData(payloadOfHandshake1);
+                    var (handshake2, _) = _nonceManager.NewNonceWithPayload(payloadOfHandshake2);
+                    await epRemote.SendBlockAsync(handshake2);
+                }
+            }
 
-            var ep = new EncryptedTcpPeer(_myPrivateKey, peerInfo.PublicKey, peer);
-            await ep.SendBlock(SHA256.HashData(data)); //ACK
+            //replace exist connection
+            if (_IncomePeers.Remove(peerId, out var exist)) exist.Disconnect();
 
             // register as income
-            //TODO: replace exist connection
-            _IncomePeers[peerId] = ep;
+            _IncomePeers[peerId] = epRemote;
 
             // start RxCycle
-            while (true) // start RxCycle
+            while (true)
             {
-                var rxb = await ep.RxBlockAsync();
-                if (rxb == null) break;
-                _eventBus.Publish(new TPM_EVT_PEER_RX(peerId, rxb));
+                var receivedBlock = await epRemote.RxBlockAsync();
+                if (receivedBlock == null) break;
+
+                var (ok, senderTimestamp, payload) = _nonceManager.CheckValidAndExtractPayload(receivedBlock);
+                if (ok != true)
+                {
+                    PublishXX(peerId, ok switch { null => XX.TimeStampSkew, false => XX.ReplayAttackDetected, _ => XX.Invalid });
+                    break;
+                }
+
+                _eventBus.Publish(new TPM_EVT_PEER_RX(peerId, senderTimestamp, payload));
             }
         }
+        catch (ConnectionAbortedException)
+        {
+            //ignore
+        }
         catch (Exception e)
         {
             _logger.LogError(e, nameof(HandleIncomingPeerAsync));
         }
 
-        if (peerId != null) _eventBus.Publish(new TPM_EVT_PEER_DX(peerId));
+        if (peerId != null)
+        {
+            _eventBus.Publish(new TPM_EVT_PEER_DX(peerId));
+            // unregister from income
+            if (_IncomePeers.Remove(peerId, out var peer))
+            {
+                peer.Disconnect();
+            }
+        }
     }
 
-    public bool SendToPeer(string peerId, byte[] block)
+    private void PublishXX(string? peerId, XX kind) => _eventBus.Publish(new TPM_EVT_PEER_XX(kind, peerId));
+
+    public bool SendToPeer(string peerId, byte[] payload)
     {
         if (_connectedPeers.TryGetValue(peerId, out var peerCon) == false && _IncomePeers.TryGetValue(peerId, out peerCon) == false) return false;
-        _ = peerCon.SendBlock(block);
+        Task.Run(async () =>
+        {
+            var (payloadAndNonce, timestamp) = _nonceManager.NewNonceWithPayload(payload);
+
+            await peerCon.SendBlockAsync(payloadAndNonce);
+
+            _eventBus.Publish(new TPM_EVT_PEER_TX(peerId, timestamp, payload));
+        });
         return true;
     }
 
-    private record PeerInfo(RSA PublicKey, string host, int port);
+    private record PeerInfo(RSA PublicKey, string Host, int Port);
 }