messageHandlers = new HashMap<>();
+ private final NodeRecordFactory nodeRecordFactory;
+
+ public DiscoveryV5MessageProcessor(NodeRecordFactory nodeRecordFactory) {
+ messageHandlers.put(MessageCode.PING, new PingHandler());
+ messageHandlers.put(MessageCode.PONG, new PongHandler());
+ messageHandlers.put(MessageCode.FINDNODE, new FindNodeHandler());
+ messageHandlers.put(MessageCode.NODES, new NodesHandler());
+ this.nodeRecordFactory = nodeRecordFactory;
+ }
+
+ @Override
+ public Protocol getSupportedIdentity() {
+ return Protocol.V5;
+ }
+
+ @Override
+ public void handleMessage(DiscoveryV5Message message, NodeSession session) {
+ MessageCode code = message.getCode();
+ MessageHandler messageHandler = messageHandlers.get(code);
+ logger.trace(() -> String.format("Handling message %s in session %s", message, session));
+ if (messageHandler == null) {
+ throw new RuntimeException("Not implemented yet");
+ }
+ messageHandler.handle(message.create(nodeRecordFactory), session);
+ }
+}
diff --git a/discovery/src/main/java/org/ethereum/beacon/discovery/Functions.java b/discovery/src/main/java/org/ethereum/beacon/discovery/Functions.java
new file mode 100644
index 000000000..f60ee7c46
--- /dev/null
+++ b/discovery/src/main/java/org/ethereum/beacon/discovery/Functions.java
@@ -0,0 +1,296 @@
+package org.ethereum.beacon.discovery;
+
+import com.google.common.base.Objects;
+import org.bouncycastle.crypto.Digest;
+import org.bouncycastle.crypto.digests.SHA256Digest;
+import org.bouncycastle.crypto.generators.HKDFBytesGenerator;
+import org.bouncycastle.crypto.params.ECDomainParameters;
+import org.bouncycastle.crypto.params.HKDFParameters;
+import org.bouncycastle.math.ec.ECPoint;
+import org.bouncycastle.util.Arrays;
+import org.ethereum.beacon.crypto.Hashes;
+import org.ethereum.beacon.util.Utils;
+import org.web3j.crypto.ECDSASignature;
+import org.web3j.crypto.ECKeyPair;
+import org.web3j.crypto.Hash;
+import org.web3j.crypto.Sign;
+import tech.pegasys.artemis.util.bytes.Bytes32;
+import tech.pegasys.artemis.util.bytes.Bytes32s;
+import tech.pegasys.artemis.util.bytes.BytesValue;
+
+import javax.crypto.Cipher;
+import javax.crypto.spec.GCMParameterSpec;
+import javax.crypto.spec.SecretKeySpec;
+import java.math.BigInteger;
+import java.security.SecureRandom;
+import java.util.Random;
+
+import static org.ethereum.beacon.util.Utils.extractBytesFromUnsignedBigInt;
+import static org.web3j.crypto.Sign.CURVE_PARAMS;
+
+/** Set of cryptography and utilities functions used in discovery */
+public class Functions {
+ public static final ECDomainParameters SECP256K1_CURVE =
+ new ECDomainParameters(
+ CURVE_PARAMS.getCurve(), CURVE_PARAMS.getG(), CURVE_PARAMS.getN(), CURVE_PARAMS.getH());
+ public static final int PUBKEY_SIZE = 64;
+ private static final int RECIPIENT_KEY_LENGTH = 16;
+ private static final int INITIATOR_KEY_LENGTH = 16;
+ private static final int AUTH_RESP_KEY_LENGTH = 16;
+ private static final int MS_IN_SECOND = 1000;
+
+ /** SHA2 (SHA256) */
+ public static Bytes32 hash(BytesValue value) {
+ return Hashes.sha256(value);
+ }
+
+ /** SHA3 (Keccak256) */
+ public static Bytes32 hashKeccak(BytesValue value) {
+ return Bytes32.wrap(Hash.sha3(value.extractArray()));
+ }
+
+ /**
+ * Creates a signature of message `x` using the given key.
+ *
+ * @param key private key
+ * @param x message, hashed
+ * @return ECDSA signature with properties merged together: r || s
+ */
+ public static BytesValue sign(BytesValue key, BytesValue x) {
+ Sign.SignatureData signatureData =
+ Sign.signMessage(x.extractArray(), ECKeyPair.create(key.extractArray()), false);
+ Bytes32 r = Bytes32.wrap(signatureData.getR());
+ Bytes32 s = Bytes32.wrap(signatureData.getS());
+ return r.concat(s);
+ }
+
+ /**
+ * Verifies that signature is made by signer
+ *
+ * @param signature Signature, ECDSA
+ * @param x message, hashed
+ * @param pubKey Public key of supposed signer, compressed, 33 bytes
+ * @return whether `signature` reflects message `x` signed with `pubkey`
+ */
+ public static boolean verifyECDSASignature(
+ BytesValue signature, BytesValue x, BytesValue pubKey) {
+ assert pubKey.size() == 33;
+ ECPoint ecPoint = Functions.publicKeyToPoint(pubKey);
+ BytesValue pubKeyUncompressed = BytesValue.wrap(ecPoint.getEncoded(false)).slice(1);
+ ECDSASignature ecdsaSignature =
+ new ECDSASignature(
+ new BigInteger(1, signature.slice(0, 32).extractArray()),
+ new BigInteger(1, signature.slice(32).extractArray()));
+ for (int recId = 0; recId < 4; ++recId) {
+ BigInteger calculatedPubKey =
+ Sign.recoverFromSignature(recId, ecdsaSignature, x.extractArray());
+ if (calculatedPubKey == null) {
+ continue;
+ }
+ if (Arrays.areEqual(
+ pubKeyUncompressed.extractArray(),
+ extractBytesFromUnsignedBigInt(calculatedPubKey, PUBKEY_SIZE))) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ /**
+ * AES-GCM encryption/authentication with the given `key`, `nonce` and additional authenticated
+ * data `ad`. Size of `key` is 16 bytes (AES-128), size of `nonce` 12 bytes.
+ */
+ public static BytesValue aesgcm_encrypt(
+ BytesValue privateKey, BytesValue nonce, BytesValue message, BytesValue aad) {
+ try {
+ Cipher cipher = Cipher.getInstance("AES/GCM/NoPadding");
+ cipher.init(
+ Cipher.ENCRYPT_MODE,
+ new SecretKeySpec(privateKey.extractArray(), "AES"),
+ new GCMParameterSpec(128, nonce.extractArray()));
+ cipher.updateAAD(aad.extractArray());
+ return BytesValue.wrap(cipher.doFinal(message.extractArray()));
+ } catch (Exception e) {
+ throw new RuntimeException("No AES/GCM cipher provider", e);
+ }
+ }
+
+ /**
+ * AES-GCM decryption of `encoded` data with the given `key`, `nonce` and additional authenticated
+ * data `ad`. Size of `key` is 16 bytes (AES-128), size of `nonce` 12 bytes.
+ */
+ public static BytesValue aesgcm_decrypt(
+ BytesValue privateKey, BytesValue nonce, BytesValue encoded, BytesValue aad) {
+ try {
+ Cipher cipher = Cipher.getInstance("AES/GCM/NoPadding");
+ cipher.init(
+ Cipher.DECRYPT_MODE,
+ new SecretKeySpec(privateKey.extractArray(), "AES"),
+ new GCMParameterSpec(128, nonce.extractArray()));
+ cipher.updateAAD(aad.extractArray());
+ return BytesValue.wrap(cipher.doFinal(encoded.extractArray()));
+ } catch (Exception e) {
+ throw new RuntimeException("No AES/GCM cipher provider", e);
+ }
+ }
+
+ /** Maps public key to point on {@link #SECP256K1_CURVE} */
+ public static ECPoint publicKeyToPoint(BytesValue pkey) {
+ byte[] destPubPointBytes;
+ if (pkey.size() == 64) { // uncompressed
+ destPubPointBytes = new byte[pkey.size() + 1];
+ destPubPointBytes[0] = 0x04; // default prefix
+ System.arraycopy(pkey.extractArray(), 0, destPubPointBytes, 1, pkey.size());
+ } else {
+ destPubPointBytes = pkey.extractArray();
+ }
+ return SECP256K1_CURVE.getCurve().decodePoint(destPubPointBytes);
+ }
+
+ /** Derives public key in SECP256K1, compressed */
+ public static BytesValue derivePublicKeyFromPrivate(BytesValue privateKey) {
+ ECKeyPair ecKeyPair = ECKeyPair.create(privateKey.extractArray());
+ final BytesValue pubKey =
+ BytesValue.wrap(
+ Utils.extractBytesFromUnsignedBigInt(ecKeyPair.getPublicKey(), PUBKEY_SIZE));
+ ECPoint ecPoint = Functions.publicKeyToPoint(pubKey);
+ return BytesValue.wrap(ecPoint.getEncoded(true));
+ }
+
+ /** Derives key agreement ECDH by multiplying private key by public */
+ public static BytesValue deriveECDHKeyAgreement(BytesValue srcPrivKey, BytesValue destPubKey) {
+ ECPoint pudDestPoint = publicKeyToPoint(destPubKey);
+ ECPoint mult = pudDestPoint.multiply(new BigInteger(1, srcPrivKey.extractArray()));
+ return BytesValue.wrap(mult.getEncoded(true));
+ }
+
+ /**
+ * The ephemeral key is used to perform Diffie-Hellman key agreement with B's static public key
+ * and the session keys are derived from it using the HKDF key derivation function.
+ *
+ *
+ * ephemeral-key = random private key
+ * ephemeral-pubkey = public key corresponding to ephemeral-key
+ * dest-pubkey = public key of B
+ * secret = agree(ephemeral-key, dest-pubkey)
+ * info = "discovery v5 key agreement" || node-id-A || node-id-B
+ * prk = HKDF-Extract(secret, id-nonce)
+ * initiator-key, recipient-key, auth-resp-key = HKDF-Expand(prk, info)
+ */
+ public static HKDFKeys hkdf_expand(
+ BytesValue srcNodeId,
+ BytesValue destNodeId,
+ BytesValue srcPrivKey,
+ BytesValue destPubKey,
+ BytesValue idNonce) {
+ BytesValue keyAgreement = deriveECDHKeyAgreement(srcPrivKey, destPubKey);
+ return hkdf_expand(srcNodeId, destNodeId, keyAgreement, idNonce);
+ }
+
+ /**
+ * {@link #hkdf_expand(BytesValue, BytesValue, BytesValue, BytesValue, BytesValue)} but with
+ * keyAgreement already derived by {@link #deriveECDHKeyAgreement(BytesValue, BytesValue)}
+ */
+ public static HKDFKeys hkdf_expand(
+ BytesValue srcNodeId, BytesValue destNodeId, BytesValue keyAgreement, BytesValue idNonce) {
+ try {
+ BytesValue info =
+ BytesValue.wrap("discovery v5 key agreement".getBytes())
+ .concat(srcNodeId)
+ .concat(destNodeId);
+ HKDFParameters hkdfParameters =
+ new HKDFParameters(
+ keyAgreement.extractArray(), idNonce.extractArray(), info.extractArray());
+ Digest digest = new SHA256Digest();
+ HKDFBytesGenerator hkdfBytesGenerator = new HKDFBytesGenerator(digest);
+ hkdfBytesGenerator.init(hkdfParameters);
+ // initiator-key || recipient-key || auth-resp-key
+ byte[] hkdfOutputBytes =
+ new byte[INITIATOR_KEY_LENGTH + RECIPIENT_KEY_LENGTH + AUTH_RESP_KEY_LENGTH];
+ hkdfBytesGenerator.generateBytes(
+ hkdfOutputBytes, 0, INITIATOR_KEY_LENGTH + RECIPIENT_KEY_LENGTH + AUTH_RESP_KEY_LENGTH);
+ BytesValue hkdfOutput = BytesValue.wrap(hkdfOutputBytes);
+ BytesValue initiatorKey = hkdfOutput.slice(0, INITIATOR_KEY_LENGTH);
+ BytesValue recipientKey = hkdfOutput.slice(INITIATOR_KEY_LENGTH, RECIPIENT_KEY_LENGTH);
+ BytesValue authRespKey = hkdfOutput.slice(INITIATOR_KEY_LENGTH + RECIPIENT_KEY_LENGTH);
+ return new HKDFKeys(initiatorKey, recipientKey, authRespKey);
+ } catch (Exception ex) {
+ throw new RuntimeException(ex);
+ }
+ }
+
+ /** Current time in seconds */
+ public static long getTime() {
+ return System.currentTimeMillis() / MS_IN_SECOND;
+ }
+
+ /** Random provider */
+ public static Random getRandom() {
+ return new SecureRandom();
+ }
+
+ /**
+ * The 'distance' between two node IDs is the bitwise XOR of the IDs, taken as the number.
+ *
+ *
distance(n₁, n₂) = n₁ XOR n₂
+ *
+ *
LogDistance is reverse of length of common prefix in bits (length - number of leftmost zeros
+ * in XOR)
+ */
+ public static int logDistance(Bytes32 nodeId1, Bytes32 nodeId2) {
+ BytesValue distance = Bytes32s.xor(nodeId1, nodeId2);
+ int logDistance = Byte.SIZE * distance.size(); // 256
+ final int maxLogDistance = logDistance;
+ for (int i = 0; i < maxLogDistance; ++i) {
+ if (distance.getHighBit(i)) {
+ break;
+ } else {
+ logDistance--;
+ }
+ }
+ return logDistance;
+ }
+
+ /**
+ * Stores set of keys derived by simple key derivation function (KDF) based on a hash-based
+ * message authentication code (HMAC)
+ */
+ public static class HKDFKeys {
+ private final BytesValue initiatorKey;
+ private final BytesValue recipientKey;
+ private final BytesValue authResponseKey;
+
+ public HKDFKeys(BytesValue initiatorKey, BytesValue recipientKey, BytesValue authResponseKey) {
+ this.initiatorKey = initiatorKey;
+ this.recipientKey = recipientKey;
+ this.authResponseKey = authResponseKey;
+ }
+
+ public BytesValue getInitiatorKey() {
+ return initiatorKey;
+ }
+
+ public BytesValue getRecipientKey() {
+ return recipientKey;
+ }
+
+ public BytesValue getAuthResponseKey() {
+ return authResponseKey;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) return true;
+ if (o == null || getClass() != o.getClass()) return false;
+ HKDFKeys hkdfKeys = (HKDFKeys) o;
+ return Objects.equal(initiatorKey, hkdfKeys.initiatorKey)
+ && Objects.equal(recipientKey, hkdfKeys.recipientKey)
+ && Objects.equal(authResponseKey, hkdfKeys.authResponseKey);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hashCode(initiatorKey, recipientKey, authResponseKey);
+ }
+ }
+}
diff --git a/discovery/src/main/java/org/ethereum/beacon/discovery/MessageProcessor.java b/discovery/src/main/java/org/ethereum/beacon/discovery/MessageProcessor.java
new file mode 100644
index 000000000..16a7bd3f7
--- /dev/null
+++ b/discovery/src/main/java/org/ethereum/beacon/discovery/MessageProcessor.java
@@ -0,0 +1,37 @@
+package org.ethereum.beacon.discovery;
+
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+import org.ethereum.beacon.discovery.message.DiscoveryMessage;
+
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * Highest level processor which knows several processors for different versions of {@link
+ * DiscoveryMessage}'s.
+ */
+public class MessageProcessor {
+ private static final Logger logger = LogManager.getLogger(MessageProcessor.class);
+ private final Map messageProcessors = new HashMap<>();
+
+ public MessageProcessor(DiscoveryMessageProcessor... messageProcessors) {
+ for (int i = 0; i < messageProcessors.length; ++i) {
+ this.messageProcessors.put(messageProcessors[i].getSupportedIdentity(), messageProcessors[i]);
+ }
+ }
+
+ public void handleIncoming(DiscoveryMessage message, NodeSession session) {
+ Protocol protocol = message.getProtocol();
+ DiscoveryMessageProcessor messageHandler = messageProcessors.get(protocol);
+ if (messageHandler == null) {
+ String error =
+ String.format(
+ "Message %s with identity %s received in session %s is not supported",
+ message, protocol, session);
+ logger.error(error);
+ throw new RuntimeException(error);
+ }
+ messageHandler.handleMessage(message, session);
+ }
+}
diff --git a/discovery/src/main/java/org/ethereum/beacon/discovery/NodeRecordInfo.java b/discovery/src/main/java/org/ethereum/beacon/discovery/NodeRecordInfo.java
new file mode 100644
index 000000000..5ec4fd91a
--- /dev/null
+++ b/discovery/src/main/java/org/ethereum/beacon/discovery/NodeRecordInfo.java
@@ -0,0 +1,101 @@
+package org.ethereum.beacon.discovery;
+
+import com.google.common.base.Objects;
+import org.ethereum.beacon.discovery.enr.NodeRecord;
+import org.ethereum.beacon.discovery.enr.NodeRecordFactory;
+import org.web3j.rlp.RlpDecoder;
+import org.web3j.rlp.RlpEncoder;
+import org.web3j.rlp.RlpList;
+import org.web3j.rlp.RlpString;
+import org.web3j.rlp.RlpType;
+import tech.pegasys.artemis.util.bytes.BytesValue;
+
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * Container for {@link NodeRecord}. Also saves all necessary data about presence of this node and
+ * last test of its availability
+ */
+public class NodeRecordInfo {
+ private final NodeRecord node;
+ private final Long lastRetry;
+ private final NodeStatus status;
+ private final Integer retry;
+
+ public NodeRecordInfo(NodeRecord node, Long lastRetry, NodeStatus status, Integer retry) {
+ this.node = node;
+ this.lastRetry = lastRetry;
+ this.status = status;
+ this.retry = retry;
+ }
+
+ public static NodeRecordInfo createDefault(NodeRecord nodeRecord) {
+ return new NodeRecordInfo(nodeRecord, -1L, NodeStatus.ACTIVE, 0);
+ }
+
+ public static NodeRecordInfo fromRlpBytes(BytesValue bytes, NodeRecordFactory nodeRecordFactory) {
+ RlpList internalList = (RlpList) RlpDecoder.decode(bytes.extractArray()).getValues().get(0);
+ return new NodeRecordInfo(
+ nodeRecordFactory.fromBytes(((RlpString) internalList.getValues().get(0)).getBytes()),
+ ((RlpString) internalList.getValues().get(1)).asPositiveBigInteger().longValue(),
+ NodeStatus.fromNumber(((RlpString) internalList.getValues().get(2)).getBytes()[0]),
+ ((RlpString) internalList.getValues().get(1)).asPositiveBigInteger().intValue());
+ }
+
+ public BytesValue toRlpBytes() {
+ List values = new ArrayList<>();
+ values.add(RlpString.create(getNode().serialize().extractArray()));
+ values.add(RlpString.create(getLastRetry()));
+ values.add(RlpString.create(getStatus().byteCode()));
+ values.add(RlpString.create(getRetry()));
+ byte[] bytes = RlpEncoder.encode(new RlpList(values));
+ return BytesValue.wrap(bytes);
+ }
+
+ public NodeRecord getNode() {
+ return (NodeRecord) node;
+ }
+
+ public Long getLastRetry() {
+ return lastRetry;
+ }
+
+ public NodeStatus getStatus() {
+ return status;
+ }
+
+ public Integer getRetry() {
+ return retry;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) return true;
+ if (o == null || getClass() != o.getClass()) return false;
+ NodeRecordInfo that = (NodeRecordInfo) o;
+ return Objects.equal(node, that.node)
+ && Objects.equal(lastRetry, that.lastRetry)
+ && status == that.status
+ && Objects.equal(retry, that.retry);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hashCode(node, lastRetry, status, retry);
+ }
+
+ @Override
+ public String toString() {
+ return "NodeRecordInfo{"
+ + "node="
+ + node
+ + ", lastRetry="
+ + lastRetry
+ + ", status="
+ + status
+ + ", retry="
+ + retry
+ + '}';
+ }
+}
diff --git a/discovery/src/main/java/org/ethereum/beacon/discovery/NodeSession.java b/discovery/src/main/java/org/ethereum/beacon/discovery/NodeSession.java
new file mode 100644
index 000000000..b1328e165
--- /dev/null
+++ b/discovery/src/main/java/org/ethereum/beacon/discovery/NodeSession.java
@@ -0,0 +1,332 @@
+package org.ethereum.beacon.discovery;
+
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+import org.ethereum.beacon.discovery.enr.NodeRecord;
+import org.ethereum.beacon.discovery.packet.Packet;
+import org.ethereum.beacon.discovery.pipeline.info.RequestInfo;
+import org.ethereum.beacon.discovery.pipeline.info.RequestInfoFactory;
+import org.ethereum.beacon.discovery.storage.AuthTagRepository;
+import org.ethereum.beacon.discovery.storage.NodeBucket;
+import org.ethereum.beacon.discovery.storage.NodeBucketStorage;
+import org.ethereum.beacon.discovery.storage.NodeTable;
+import org.ethereum.beacon.discovery.task.TaskOptions;
+import org.ethereum.beacon.discovery.task.TaskType;
+import org.ethereum.beacon.util.ExpirationScheduler;
+import tech.pegasys.artemis.util.bytes.Bytes32;
+import tech.pegasys.artemis.util.bytes.BytesValue;
+
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Random;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Consumer;
+
+import static org.ethereum.beacon.discovery.task.TaskStatus.AWAIT;
+
+/**
+ * Stores session status and all keys for discovery message exchange between us, `homeNode` and the
+ * other `node`
+ */
+public class NodeSession {
+ public static final int NONCE_SIZE = 12;
+ public static final int REQUEST_ID_SIZE = 8;
+ private static final Logger logger = LogManager.getLogger(NodeSession.class);
+ private static final int CLEANUP_DELAY_SECONDS = 60;
+ private final NodeRecord homeNodeRecord;
+ private final Bytes32 homeNodeId;
+ private final AuthTagRepository authTagRepo;
+ private final NodeTable nodeTable;
+ private final NodeBucketStorage nodeBucketStorage;
+ private final Consumer outgoing;
+ private final Random rnd;
+ private NodeRecord nodeRecord;
+ private SessionStatus status = SessionStatus.INITIAL;
+ private Bytes32 idNonce;
+ private BytesValue initiatorKey;
+ private BytesValue recipientKey;
+ private Map requestIdStatuses = new ConcurrentHashMap<>();
+ private ExpirationScheduler requestExpirationScheduler =
+ new ExpirationScheduler<>(CLEANUP_DELAY_SECONDS, TimeUnit.SECONDS);
+ private CompletableFuture completableFuture = null;
+ private BytesValue staticNodeKey;
+
+ public NodeSession(
+ NodeRecord nodeRecord,
+ NodeRecord homeNodeRecord,
+ BytesValue staticNodeKey,
+ NodeTable nodeTable,
+ NodeBucketStorage nodeBucketStorage,
+ AuthTagRepository authTagRepo,
+ Consumer outgoing,
+ Random rnd) {
+ this.nodeRecord = nodeRecord;
+ this.outgoing = outgoing;
+ this.authTagRepo = authTagRepo;
+ this.nodeTable = nodeTable;
+ this.nodeBucketStorage = nodeBucketStorage;
+ this.homeNodeRecord = homeNodeRecord;
+ this.staticNodeKey = staticNodeKey;
+ this.homeNodeId = homeNodeRecord.getNodeId();
+ this.rnd = rnd;
+ }
+
+ public NodeRecord getNodeRecord() {
+ return nodeRecord;
+ }
+
+ public synchronized void updateNodeRecord(NodeRecord nodeRecord) {
+ logger.trace(
+ () ->
+ String.format(
+ "NodeRecord updated from %s to %s in session %s",
+ this.nodeRecord, nodeRecord, this));
+ this.nodeRecord = nodeRecord;
+ }
+
+ private void completeConnectFuture() {
+ if (completableFuture != null) {
+ completableFuture.complete(null);
+ completableFuture = null;
+ }
+ }
+
+ public void sendOutgoing(Packet packet) {
+ logger.trace(() -> String.format("Sending outgoing packet %s in session %s", packet, this));
+ outgoing.accept(packet);
+ }
+
+ /**
+ * Creates object with request information: requestId etc, RequestInfo, designed to maintain
+ * request status and its changes. Also stores info in session repository to track related
+ * messages.
+ *
+ * The value selected as request ID must allow for concurrent conversations. Using a timestamp
+ * can result in parallel conversations with the same id, so this should be avoided. Request IDs
+ * also prevent replay of responses. Using a simple counter would be fine if the implementation
+ * could ensure that restarts or even re-installs would increment the counter based on previously
+ * saved state in all circumstances. The easiest to implement is a random number.
+ *
+ * @param taskType Type of task, clarifies starting and reply message types
+ * @param taskOptions Task options
+ * @param future Future to be fired when task is successfully completed or exceptionally break
+ * when its failed
+ * @return info bundle.
+ */
+ public synchronized RequestInfo createNextRequest(
+ TaskType taskType, TaskOptions taskOptions, CompletableFuture future) {
+ byte[] requestId = new byte[REQUEST_ID_SIZE];
+ rnd.nextBytes(requestId);
+ BytesValue wrappedId = BytesValue.wrap(requestId);
+ if (taskOptions.isLivenessUpdate()) {
+ future.whenComplete(
+ (aVoid, throwable) -> {
+ if (throwable == null) {
+ updateLiveness();
+ }
+ });
+ }
+ RequestInfo requestInfo = RequestInfoFactory.create(taskType, wrappedId, taskOptions, future);
+ requestIdStatuses.put(wrappedId, requestInfo);
+ requestExpirationScheduler.put(
+ wrappedId,
+ new Runnable() {
+ @Override
+ public void run() {
+ logger.debug(
+ () ->
+ String.format(
+ "Request %s expired for id %s in session %s: no reply",
+ requestInfo, wrappedId, this));
+ requestIdStatuses.remove(wrappedId);
+ }
+ });
+ return requestInfo;
+ }
+
+ /** Updates request info. Thread-safe. */
+ public synchronized void updateRequestInfo(BytesValue requestId, RequestInfo newRequestInfo) {
+ RequestInfo oldRequestInfo = requestIdStatuses.remove(requestId);
+ if (oldRequestInfo == null) {
+ logger.debug(
+ () ->
+ String.format(
+ "An attempt to update requestId %s in session %s which does not exist",
+ requestId, this));
+ return;
+ }
+ requestIdStatuses.put(requestId, newRequestInfo);
+ requestExpirationScheduler.put(
+ requestId,
+ new Runnable() {
+ @Override
+ public void run() {
+ logger.debug(
+ String.format(
+ "Request %s expired for id %s in session %s: no reply",
+ newRequestInfo, requestId, this));
+ requestIdStatuses.remove(requestId);
+ }
+ });
+ }
+
+ public synchronized void cancelAllRequests(String message) {
+ logger.debug(() -> String.format("Cancelling all requests in session %s", this));
+ Set requestIdsCopy = new HashSet<>(requestIdStatuses.keySet());
+ requestIdsCopy.forEach(
+ requestId -> {
+ RequestInfo requestInfo = clearRequestId(requestId);
+ requestInfo
+ .getFuture()
+ .completeExceptionally(
+ new RuntimeException(
+ String.format(
+ "Request %s cancelled due to reason: %s", requestInfo, message)));
+ });
+ }
+
+ /** Generates random nonce of {@link #NONCE_SIZE} size */
+ public synchronized BytesValue generateNonce() {
+ byte[] nonce = new byte[NONCE_SIZE];
+ rnd.nextBytes(nonce);
+ return BytesValue.wrap(nonce);
+ }
+
+ /** If true indicates that handshake is complete */
+ public synchronized boolean isAuthenticated() {
+ return SessionStatus.AUTHENTICATED.equals(status);
+ }
+
+ /** Resets stored authTags for this session making them obsolete */
+ public void cleanup() {
+ authTagRepo.expire(this);
+ }
+
+ public Optional getAuthTag() {
+ return authTagRepo.getTag(this);
+ }
+
+ public void setAuthTag(BytesValue authTag) {
+ authTagRepo.put(authTag, this);
+ }
+
+ public Bytes32 getHomeNodeId() {
+ return homeNodeId;
+ }
+
+ /** @return initiator key, also known as write key */
+ public BytesValue getInitiatorKey() {
+ return initiatorKey;
+ }
+
+ public void setInitiatorKey(BytesValue initiatorKey) {
+ this.initiatorKey = initiatorKey;
+ }
+
+ /** @return recipient key, also known as read key */
+ public BytesValue getRecipientKey() {
+ return recipientKey;
+ }
+
+ public void setRecipientKey(BytesValue recipientKey) {
+ this.recipientKey = recipientKey;
+ }
+
+ public synchronized void clearRequestId(BytesValue requestId, TaskType taskType) {
+ RequestInfo requestInfo = clearRequestId(requestId);
+ requestInfo.getFuture().complete(null);
+ assert taskType.equals(requestInfo.getTaskType());
+ }
+
+ /** Updates nodeRecord {@link NodeStatus} to ACTIVE of the node associated with this session */
+ public synchronized void updateLiveness() {
+ NodeRecordInfo nodeRecordInfo =
+ new NodeRecordInfo(getNodeRecord(), Functions.getTime(), NodeStatus.ACTIVE, 0);
+ nodeTable.save(nodeRecordInfo);
+ nodeBucketStorage.put(nodeRecordInfo);
+ }
+
+ private synchronized RequestInfo clearRequestId(BytesValue requestId) {
+ RequestInfo requestInfo = requestIdStatuses.remove(requestId);
+ requestExpirationScheduler.cancel(requestId);
+ return requestInfo;
+ }
+
+ public synchronized Optional getRequestId(BytesValue requestId) {
+ RequestInfo requestInfo = requestIdStatuses.get(requestId);
+ return requestId == null ? Optional.empty() : Optional.of(requestInfo);
+ }
+
+ /**
+ * Returns any queued {@link RequestInfo} which was not started because session is not
+ * authenticated
+ */
+ public synchronized Optional getFirstAwaitRequestInfo() {
+ return requestIdStatuses.values().stream()
+ .filter(requestInfo -> AWAIT.equals(requestInfo.getTaskStatus()))
+ .findFirst();
+ }
+
+ public NodeTable getNodeTable() {
+ return nodeTable;
+ }
+
+ public void putRecordInBucket(NodeRecordInfo nodeRecordInfo) {
+ nodeBucketStorage.put(nodeRecordInfo);
+ }
+
+ public Optional getBucket(int index) {
+ return nodeBucketStorage.get(index);
+ }
+
+ public synchronized Bytes32 getIdNonce() {
+ return idNonce;
+ }
+
+ public synchronized void setIdNonce(Bytes32 idNonce) {
+ this.idNonce = idNonce;
+ }
+
+ public NodeRecord getHomeNodeRecord() {
+ return homeNodeRecord;
+ }
+
+ @Override
+ public String toString() {
+ return "NodeSession{"
+ + "nodeRecord="
+ + nodeRecord
+ + ", homeNodeId="
+ + homeNodeId
+ + ", status="
+ + status
+ + '}';
+ }
+
+ public synchronized SessionStatus getStatus() {
+ return status;
+ }
+
+ public synchronized void setStatus(SessionStatus newStatus) {
+ logger.debug(
+ () ->
+ String.format(
+ "Switching status of node %s from %s to %s", nodeRecord, status, newStatus));
+ this.status = newStatus;
+ }
+
+ public BytesValue getStaticNodeKey() {
+ return staticNodeKey;
+ }
+
+ public enum SessionStatus {
+ INITIAL, // other side is trying to connect, or we are initiating (before random packet is sent
+ WHOAREYOU_SENT, // other side is initiator, we've sent whoareyou in response
+ RANDOM_PACKET_SENT, // our node is initiator, we've sent random packet
+ AUTHENTICATED
+ }
+}
diff --git a/discovery/src/main/java/org/ethereum/beacon/discovery/NodeStatus.java b/discovery/src/main/java/org/ethereum/beacon/discovery/NodeStatus.java
new file mode 100644
index 000000000..c96dd707d
--- /dev/null
+++ b/discovery/src/main/java/org/ethereum/beacon/discovery/NodeStatus.java
@@ -0,0 +1,33 @@
+package org.ethereum.beacon.discovery;
+
+import java.util.HashMap;
+import java.util.Map;
+
+/** Status of {@link org.ethereum.beacon.discovery.enr.NodeRecord} */
+public enum NodeStatus {
+ ACTIVE(0x01), // Alive
+ SLEEP(0x02), // Didn't answer last time(s)
+ DEAD(0x03); // Didnt' answer for a long time
+
+ private static final Map codeMap = new HashMap<>();
+
+ static {
+ for (NodeStatus type : NodeStatus.values()) {
+ codeMap.put(type.code, type);
+ }
+ }
+
+ private int code;
+
+ NodeStatus(int code) {
+ this.code = code;
+ }
+
+ public static NodeStatus fromNumber(int i) {
+ return codeMap.get(i);
+ }
+
+ public byte byteCode() {
+ return (byte) code;
+ }
+}
diff --git a/discovery/src/main/java/org/ethereum/beacon/discovery/Protocol.java b/discovery/src/main/java/org/ethereum/beacon/discovery/Protocol.java
new file mode 100644
index 000000000..beaf30f9a
--- /dev/null
+++ b/discovery/src/main/java/org/ethereum/beacon/discovery/Protocol.java
@@ -0,0 +1,32 @@
+package org.ethereum.beacon.discovery;
+
+import java.util.HashMap;
+import java.util.Map;
+
+/** Discovery protocol versions */
+public enum Protocol {
+ V4("v4"),
+ V5("v5");
+
+ private static final Map nameMap = new HashMap<>();
+
+ static {
+ for (Protocol scheme : Protocol.values()) {
+ nameMap.put(scheme.name, scheme);
+ }
+ }
+
+ private String name;
+
+ private Protocol(String name) {
+ this.name = name;
+ }
+
+ public static Protocol fromString(String name) {
+ return nameMap.get(name);
+ }
+
+ public String stringName() {
+ return name;
+ }
+}
diff --git a/discovery/src/main/java/org/ethereum/beacon/discovery/RlpUtil.java b/discovery/src/main/java/org/ethereum/beacon/discovery/RlpUtil.java
new file mode 100644
index 000000000..b56190ac0
--- /dev/null
+++ b/discovery/src/main/java/org/ethereum/beacon/discovery/RlpUtil.java
@@ -0,0 +1,102 @@
+package org.ethereum.beacon.discovery;
+
+import org.ethereum.beacon.discovery.enr.IdentitySchema;
+import org.javatuples.Pair;
+import org.web3j.rlp.RlpDecoder;
+import org.web3j.rlp.RlpList;
+import org.web3j.rlp.RlpString;
+import tech.pegasys.artemis.util.bytes.Bytes8;
+import tech.pegasys.artemis.util.bytes.BytesValue;
+import tech.pegasys.artemis.util.uint.UInt64;
+
+import java.math.BigInteger;
+import java.util.function.Function;
+
+import static org.web3j.rlp.RlpDecoder.OFFSET_LONG_LIST;
+import static org.web3j.rlp.RlpDecoder.OFFSET_SHORT_LIST;
+
+/**
+ * Handy utilities used for RLP encoding and decoding and not fulfilled by {@link
+ * org.web3j.rlp.RlpEncoder} and {@link RlpDecoder}
+ */
+public class RlpUtil {
+ /**
+ * Calculates length of list beginning from the start of the data. So, there could everything else
+ * after first list in data, method helps to cut data in this case.
+ */
+ private static int calcListLen(BytesValue data) {
+ int prefix = data.get(0) & 0xFF;
+ int prefixAddon = 1;
+ if (prefix >= OFFSET_SHORT_LIST && prefix <= OFFSET_LONG_LIST) {
+
+ // 4. the data is a list if the range of the
+ // first byte is [0xc0, 0xf7], and the concatenation of
+ // the RLP encodings of all items of the list which the
+ // total payload is equal to the first byte minus 0xc0 follows the first byte;
+
+ byte listLen = (byte) (prefix - OFFSET_SHORT_LIST);
+ return listLen & 0xFF + prefixAddon;
+ } else if (prefix > OFFSET_LONG_LIST) {
+
+ // 5. the data is a list if the range of the
+ // first byte is [0xf8, 0xff], and the total payload of the
+ // list which length is equal to the
+ // first byte minus 0xf7 follows the first byte,
+ // and the concatenation of the RLP encodings of all items of
+ // the list follows the total payload of the list;
+
+ int lenOfListLen = (prefix - OFFSET_LONG_LIST) & 0xFF;
+ prefixAddon += lenOfListLen;
+ return UInt64.fromBytesBigEndian(Bytes8.leftPad(data.slice(1, lenOfListLen & 0xFF)))
+ .intValue()
+ + prefixAddon;
+ } else {
+ throw new RuntimeException("Not a start of RLP list!!");
+ }
+ }
+
+ /**
+ * @return first rlp list in provided data, plus remaining data starting from the end of this list
+ */
+ public static Pair decodeFirstList(BytesValue data) {
+ int len = RlpUtil.calcListLen(data);
+ return Pair.with(RlpDecoder.decode(data.slice(0, len).extractArray()), data.slice(len));
+ }
+
+ /**
+ * Encodes object to {@link RlpString}. Supports numbers, {@link BytesValue} etc.
+ *
+ * @throws RuntimeException with errorMessageFunction applied with `object` when encoding is not
+ * possible
+ */
+ public static RlpString encode(Object object, Function errorMessageFunction) {
+ if (object instanceof BytesValue) {
+ return fromBytesValue((BytesValue) object);
+ } else if (object instanceof Number) {
+ return fromNumber((Number) object);
+ } else if (object == null) {
+ return RlpString.create(new byte[0]);
+ } else if (object instanceof IdentitySchema) {
+ return RlpString.create(((IdentitySchema) object).stringName());
+ } else {
+ throw new RuntimeException(errorMessageFunction.apply(object));
+ }
+ }
+
+ private static RlpString fromNumber(Number number) {
+ if (number instanceof BigInteger) {
+ return RlpString.create((BigInteger) number);
+ } else if (number instanceof Long) {
+ return RlpString.create((Long) number);
+ } else if (number instanceof Integer) {
+ return RlpString.create((Integer) number);
+ } else {
+ throw new RuntimeException(
+ String.format("Couldn't serialize number %s : no serializer found.", number));
+ }
+ }
+
+ private static RlpString fromBytesValue(BytesValue bytes) {
+ return RlpString.create(bytes.extractArray());
+ }
+}
diff --git a/discovery/src/main/java/org/ethereum/beacon/discovery/enr/EnrField.java b/discovery/src/main/java/org/ethereum/beacon/discovery/enr/EnrField.java
new file mode 100644
index 000000000..291539445
--- /dev/null
+++ b/discovery/src/main/java/org/ethereum/beacon/discovery/enr/EnrField.java
@@ -0,0 +1,21 @@
+package org.ethereum.beacon.discovery.enr;
+
+/**
+ * Fields of Ethereum Node Record
+ */
+public interface EnrField {
+ // Schema id
+ String ID = "id";
+ // IPv4 address
+ String IP_V4 = "ip";
+ // TCP port, integer
+ String TCP_V4 = "tcp";
+ // UDP port, integer
+ String UDP_V4 = "udp";
+ // IPv6 address
+ String IP_V6 = "ip6";
+ // IPv6-specific TCP port
+ String TCP_V6 = "tcp6";
+ // IPv6-specific UDP port
+ String UDP_V6 = "udp6";
+}
diff --git a/discovery/src/main/java/org/ethereum/beacon/discovery/enr/EnrFieldInterpreter.java b/discovery/src/main/java/org/ethereum/beacon/discovery/enr/EnrFieldInterpreter.java
new file mode 100644
index 000000000..507e271a6
--- /dev/null
+++ b/discovery/src/main/java/org/ethereum/beacon/discovery/enr/EnrFieldInterpreter.java
@@ -0,0 +1,13 @@
+package org.ethereum.beacon.discovery.enr;
+
+import org.web3j.rlp.RlpString;
+import org.web3j.rlp.RlpType;
+
+/**
+ * Encoder/decoder for fields of ethereum node record
+ */
+public interface EnrFieldInterpreter {
+ Object decode(String key, RlpString rlpString);
+
+ RlpType encode(String key, Object object);
+}
diff --git a/discovery/src/main/java/org/ethereum/beacon/discovery/enr/EnrFieldInterpreterV4.java b/discovery/src/main/java/org/ethereum/beacon/discovery/enr/EnrFieldInterpreterV4.java
new file mode 100644
index 000000000..b6b78267f
--- /dev/null
+++ b/discovery/src/main/java/org/ethereum/beacon/discovery/enr/EnrFieldInterpreterV4.java
@@ -0,0 +1,51 @@
+package org.ethereum.beacon.discovery.enr;
+
+import org.ethereum.beacon.discovery.RlpUtil;
+import org.web3j.rlp.RlpString;
+import org.web3j.rlp.RlpType;
+import tech.pegasys.artemis.util.bytes.Bytes16;
+import tech.pegasys.artemis.util.bytes.Bytes4;
+import tech.pegasys.artemis.util.bytes.BytesValue;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.function.Function;
+
+public class EnrFieldInterpreterV4 implements EnrFieldInterpreter {
+ public static EnrFieldInterpreterV4 DEFAULT = new EnrFieldInterpreterV4();
+
+ private Map> fieldDecoders = new HashMap<>();
+
+ public EnrFieldInterpreterV4() {
+ fieldDecoders.put(
+ EnrFieldV4.PKEY_SECP256K1, rlpString -> BytesValue.wrap(rlpString.getBytes()));
+ fieldDecoders.put(
+ EnrField.ID, rlpString -> IdentitySchema.fromString(new String(rlpString.getBytes())));
+ fieldDecoders.put(
+ EnrField.IP_V4, rlpString -> Bytes4.wrap(BytesValue.wrap(rlpString.getBytes()), 0));
+ fieldDecoders.put(EnrField.TCP_V4, rlpString -> rlpString.asPositiveBigInteger().intValue());
+ fieldDecoders.put(EnrField.UDP_V4, rlpString -> rlpString.asPositiveBigInteger().intValue());
+ fieldDecoders.put(
+ EnrField.IP_V6, rlpString -> Bytes16.wrap(BytesValue.wrap(rlpString.getBytes()), 0));
+ fieldDecoders.put(EnrField.TCP_V6, rlpString -> rlpString.asPositiveBigInteger().intValue());
+ fieldDecoders.put(EnrField.UDP_V6, rlpString -> rlpString.asPositiveBigInteger().intValue());
+ }
+
+ @Override
+ public Object decode(String key, RlpString rlpString) {
+ Function fieldDecoder = fieldDecoders.get(key);
+ if (fieldDecoder == null) {
+ throw new RuntimeException(String.format("No decoder found for field `%s`", key));
+ }
+ return fieldDecoder.apply(rlpString);
+ }
+
+ @Override
+ public RlpType encode(String key, Object object) {
+ return RlpUtil.encode(
+ object,
+ o ->
+ String.format(
+ "Couldn't encode field %s with value %s: no serializer found.", key, object));
+ }
+}
diff --git a/discovery/src/main/java/org/ethereum/beacon/discovery/enr/EnrFieldV4.java b/discovery/src/main/java/org/ethereum/beacon/discovery/enr/EnrFieldV4.java
new file mode 100644
index 000000000..b216975b7
--- /dev/null
+++ b/discovery/src/main/java/org/ethereum/beacon/discovery/enr/EnrFieldV4.java
@@ -0,0 +1,10 @@
+package org.ethereum.beacon.discovery.enr;
+
+/**
+ * Fields of Ethereum Node Record V4 as defined by https://eips.ethereum.org/EIPS/eip-778
+ */
+public interface EnrFieldV4 extends EnrField {
+ // Compressed secp256k1 public key, 33 bytes
+ String PKEY_SECP256K1 = "secp256k1";
+}
diff --git a/discovery/src/main/java/org/ethereum/beacon/discovery/enr/IdentitySchema.java b/discovery/src/main/java/org/ethereum/beacon/discovery/enr/IdentitySchema.java
new file mode 100644
index 000000000..5fce90f49
--- /dev/null
+++ b/discovery/src/main/java/org/ethereum/beacon/discovery/enr/IdentitySchema.java
@@ -0,0 +1,31 @@
+package org.ethereum.beacon.discovery.enr;
+
+import java.util.HashMap;
+import java.util.Map;
+
+/** Available identity schemas of Ethereum {@link NodeRecord} signature */
+public enum IdentitySchema {
+ V4("v4");
+
+ private static final Map nameMap = new HashMap<>();
+
+ static {
+ for (IdentitySchema scheme : IdentitySchema.values()) {
+ nameMap.put(scheme.name, scheme);
+ }
+ }
+
+ private String name;
+
+ private IdentitySchema(String name) {
+ this.name = name;
+ }
+
+ public static IdentitySchema fromString(String name) {
+ return nameMap.get(name);
+ }
+
+ public String stringName() {
+ return name;
+ }
+}
diff --git a/discovery/src/main/java/org/ethereum/beacon/discovery/enr/IdentitySchemaInterpreter.java b/discovery/src/main/java/org/ethereum/beacon/discovery/enr/IdentitySchemaInterpreter.java
new file mode 100644
index 000000000..f1a685b20
--- /dev/null
+++ b/discovery/src/main/java/org/ethereum/beacon/discovery/enr/IdentitySchemaInterpreter.java
@@ -0,0 +1,30 @@
+package org.ethereum.beacon.discovery.enr;
+
+import tech.pegasys.artemis.util.bytes.Bytes32;
+
+/**
+ * Interprets identity schema of ethereum node record:
+ *
+ *
+ * - derives node id from node record
+ *
- >signs node record
+ *
- verifies signature of node record
+ *
+ */
+public interface IdentitySchemaInterpreter {
+ /** Returns supported scheme */
+ IdentitySchema getScheme();
+
+ /* Signs nodeRecord, modifying it */
+ void sign(NodeRecord nodeRecord, Object signOptions);
+
+ /** Verifies that `nodeRecord` is of scheme implementation */
+ default void verify(NodeRecord nodeRecord) {
+ if (!nodeRecord.getIdentityScheme().equals(getScheme())) {
+ throw new RuntimeException("Interpreter and node record schemes do not match!");
+ }
+ }
+
+ /** Delivers nodeId according to identity scheme scheme */
+ Bytes32 getNodeId(NodeRecord nodeRecord);
+}
diff --git a/discovery/src/main/java/org/ethereum/beacon/discovery/enr/IdentitySchemaV4Interpreter.java b/discovery/src/main/java/org/ethereum/beacon/discovery/enr/IdentitySchemaV4Interpreter.java
new file mode 100644
index 000000000..b26f5e5c1
--- /dev/null
+++ b/discovery/src/main/java/org/ethereum/beacon/discovery/enr/IdentitySchemaV4Interpreter.java
@@ -0,0 +1,50 @@
+package org.ethereum.beacon.discovery.enr;
+
+import org.bouncycastle.math.ec.ECPoint;
+import org.ethereum.beacon.discovery.Functions;
+import org.ethereum.beacon.util.Utils;
+import tech.pegasys.artemis.util.bytes.Bytes32;
+import tech.pegasys.artemis.util.bytes.BytesValue;
+
+public class IdentitySchemaV4Interpreter implements IdentitySchemaInterpreter {
+ @Override
+ public void verify(NodeRecord nodeRecord) {
+ IdentitySchemaInterpreter.super.verify(nodeRecord);
+ if (nodeRecord.get(EnrFieldV4.PKEY_SECP256K1) == null) {
+ throw new RuntimeException(
+ String.format(
+ "Field %s not exists but required for scheme %s",
+ EnrFieldV4.PKEY_SECP256K1, getScheme()));
+ }
+ BytesValue pubKey = (BytesValue) nodeRecord.get(EnrFieldV4.PKEY_SECP256K1); // compressed
+ assert Functions.verifyECDSASignature(
+ nodeRecord.getSignature(), Functions.hashKeccak(nodeRecord.serializeNoSignature()), pubKey);
+ }
+
+ @Override
+ public IdentitySchema getScheme() {
+ return IdentitySchema.V4;
+ }
+
+ @Override
+ public Bytes32 getNodeId(NodeRecord nodeRecord) {
+ verify(nodeRecord);
+ BytesValue pkey = (BytesValue) nodeRecord.getKey(EnrFieldV4.PKEY_SECP256K1);
+ ECPoint pudDestPoint = Functions.publicKeyToPoint(pkey);
+ BytesValue xPart =
+ Bytes32.wrap(
+ Utils.extractBytesFromUnsignedBigInt(pudDestPoint.getXCoord().toBigInteger(), 32));
+ BytesValue yPart =
+ Bytes32.wrap(
+ Utils.extractBytesFromUnsignedBigInt(pudDestPoint.getYCoord().toBigInteger(), 32));
+ return Functions.hashKeccak(xPart.concat(yPart));
+ }
+
+ @Override
+ public void sign(NodeRecord nodeRecord, Object signOptions) {
+ BytesValue privateKey = (BytesValue) signOptions;
+ BytesValue signature =
+ Functions.sign(privateKey, Functions.hashKeccak(nodeRecord.serializeNoSignature()));
+ nodeRecord.setSignature(signature);
+ }
+}
diff --git a/discovery/src/main/java/org/ethereum/beacon/discovery/enr/NodeRecord.java b/discovery/src/main/java/org/ethereum/beacon/discovery/enr/NodeRecord.java
new file mode 100644
index 000000000..7ef5c7170
--- /dev/null
+++ b/discovery/src/main/java/org/ethereum/beacon/discovery/enr/NodeRecord.java
@@ -0,0 +1,198 @@
+package org.ethereum.beacon.discovery.enr;
+
+import com.google.common.base.Objects;
+import org.javatuples.Pair;
+import org.web3j.rlp.RlpEncoder;
+import org.web3j.rlp.RlpList;
+import org.web3j.rlp.RlpString;
+import org.web3j.rlp.RlpType;
+import tech.pegasys.artemis.util.bytes.Bytes32;
+import tech.pegasys.artemis.util.bytes.Bytes96;
+import tech.pegasys.artemis.util.bytes.BytesValue;
+import tech.pegasys.artemis.util.uint.UInt64;
+
+import java.util.ArrayList;
+import java.util.Base64;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+/**
+ * Ethereum Node Record V4
+ *
+ * Node record as described in EIP-778
+ */
+public class NodeRecord {
+ /**
+ * The canonical encoding of a node record is an RLP list of [signature, seq, k, v, ...]. The
+ * maximum encoded size of a node record is 300 bytes. Implementations should reject records
+ * larger than this size.
+ */
+ public static final int MAX_ENCODED_SIZE = 300;
+ private static final EnrFieldInterpreter enrFieldInterpreter = EnrFieldInterpreterV4.DEFAULT;
+ private final UInt64 seq;
+ // Signature
+ private BytesValue signature;
+ // optional fields
+ private Map fields = new HashMap<>();
+ private IdentitySchemaInterpreter identitySchemaInterpreter;
+
+ private NodeRecord(
+ IdentitySchemaInterpreter identitySchemaInterpreter, UInt64 seq, BytesValue signature) {
+ this.seq = seq;
+ this.signature = signature;
+ this.identitySchemaInterpreter = identitySchemaInterpreter;
+ }
+
+ private NodeRecord(
+ IdentitySchemaInterpreter identitySchemaInterpreter, UInt64 seq) {
+ this.seq = seq;
+ this.signature = Bytes96.ZERO;
+ this.identitySchemaInterpreter = identitySchemaInterpreter;
+ }
+
+ public static NodeRecord fromValues(
+ IdentitySchemaInterpreter identitySchemaInterpreter,
+ UInt64 seq,
+ List> fieldKeyPairs) {
+ NodeRecord nodeRecord = new NodeRecord(identitySchemaInterpreter, seq);
+ fieldKeyPairs.forEach(objects -> nodeRecord.set(objects.getValue0(), objects.getValue1()));
+ return nodeRecord;
+ }
+
+ public static NodeRecord fromRawFields(
+ IdentitySchemaInterpreter identitySchemaInterpreter,
+ UInt64 seq,
+ BytesValue signature,
+ List rawFields) {
+ NodeRecord nodeRecord = new NodeRecord(identitySchemaInterpreter, seq, signature);
+ for (int i = 0; i < rawFields.size(); i += 2) {
+ String key = new String(((RlpString) rawFields.get(i)).getBytes());
+ nodeRecord.set(key, enrFieldInterpreter.decode(key, (RlpString) rawFields.get(i + 1)));
+ }
+ return nodeRecord;
+ }
+
+ public String asBase64() {
+ return new String(Base64.getUrlEncoder().encode(serialize().extractArray()));
+ }
+
+ public IdentitySchema getIdentityScheme() {
+ return identitySchemaInterpreter.getScheme();
+ }
+
+ public void set(String key, Object value) {
+ fields.put(key, value);
+ }
+
+ public Object get(String key) {
+ return fields.get(key);
+ }
+
+ public UInt64 getSeq() {
+ return seq;
+ }
+
+ public BytesValue getSignature() {
+ return signature;
+ }
+
+ public void setSignature(BytesValue signature) {
+ this.signature = signature;
+ }
+
+ public Set getKeys() {
+ return new HashSet<>(fields.keySet());
+ }
+
+ public Object getKey(String key) {
+ return fields.get(key);
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) return true;
+ if (o == null || getClass() != o.getClass()) return false;
+ NodeRecord that = (NodeRecord) o;
+ return Objects.equal(seq, that.seq)
+ && Objects.equal(signature, that.signature)
+ && Objects.equal(fields, that.fields);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hashCode(seq, signature, fields);
+ }
+
+ public void verify() {
+ identitySchemaInterpreter.verify(this);
+ }
+
+ public void sign(Object signOptions) {
+ identitySchemaInterpreter.sign(this, signOptions);
+ }
+
+ public RlpList asRlp() {
+ return asRlpImpl(true);
+ }
+
+ public RlpList asRlpNoSignature() {
+ return asRlpImpl(false);
+ }
+
+ private RlpList asRlpImpl(boolean withSignature) {
+ assert getSeq() != null;
+ // content = [seq, k, v, ...]
+ // signature = sign(content)
+ // record = [signature, seq, k, v, ...]
+ List values = new ArrayList<>();
+ if (withSignature) {
+ values.add(RlpString.create(getSignature().extractArray()));
+ }
+ values.add(RlpString.create(getSeq().toBI()));
+ List keySortedList = fields.keySet().stream().sorted().collect(Collectors.toList());
+ for (String key : keySortedList) {
+ if (fields.get(key) == null) {
+ continue;
+ }
+ values.add(RlpString.create(key));
+ values.add(enrFieldInterpreter.encode(key, fields.get(key)));
+ }
+
+ return new RlpList(values);
+ }
+
+ public BytesValue serialize() {
+ return serializeImpl(true);
+ }
+
+ public BytesValue serializeNoSignature() {
+ return serializeImpl(false);
+ }
+
+ private BytesValue serializeImpl(boolean withSignature) {
+ RlpType rlpRecord = withSignature ? asRlp() : asRlpNoSignature();
+ byte[] bytes = RlpEncoder.encode(rlpRecord);
+ assert bytes.length <= MAX_ENCODED_SIZE;
+ return BytesValue.wrap(bytes);
+ }
+
+ public Bytes32 getNodeId() {
+ return identitySchemaInterpreter.getNodeId(this);
+ }
+
+ @Override
+ public String toString() {
+ return "NodeRecordV4{"
+ + "publicKey="
+ + fields.get(EnrFieldV4.PKEY_SECP256K1)
+ + ", ipV4address="
+ + fields.get(EnrFieldV4.IP_V4)
+ + ", udpPort="
+ + fields.get(EnrFieldV4.UDP_V4)
+ + '}';
+ }
+}
diff --git a/discovery/src/main/java/org/ethereum/beacon/discovery/enr/NodeRecordFactory.java b/discovery/src/main/java/org/ethereum/beacon/discovery/enr/NodeRecordFactory.java
new file mode 100644
index 000000000..22481955f
--- /dev/null
+++ b/discovery/src/main/java/org/ethereum/beacon/discovery/enr/NodeRecordFactory.java
@@ -0,0 +1,116 @@
+package org.ethereum.beacon.discovery.enr;
+
+import org.javatuples.Pair;
+import org.web3j.rlp.RlpDecoder;
+import org.web3j.rlp.RlpList;
+import org.web3j.rlp.RlpString;
+import org.web3j.rlp.RlpType;
+import tech.pegasys.artemis.util.bytes.Bytes8;
+import tech.pegasys.artemis.util.bytes.BytesValue;
+import tech.pegasys.artemis.util.uint.UInt64;
+
+import java.util.Arrays;
+import java.util.Base64;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+public class NodeRecordFactory {
+ public static final NodeRecordFactory DEFAULT =
+ new NodeRecordFactory(new IdentitySchemaV4Interpreter());
+ Map interpreters = new HashMap<>();
+
+ public NodeRecordFactory(IdentitySchemaInterpreter... identitySchemaInterpreters) {
+ for (IdentitySchemaInterpreter identitySchemaInterpreter : identitySchemaInterpreters) {
+ interpreters.put(identitySchemaInterpreter.getScheme(), identitySchemaInterpreter);
+ }
+ }
+
+ @SafeVarargs
+ public final NodeRecord createFromValues(UInt64 seq, Pair... fieldKeyPairs) {
+ return createFromValues(seq, Arrays.asList(fieldKeyPairs));
+ }
+
+ public NodeRecord createFromValues(UInt64 seq, List> fieldKeyPairs) {
+ Pair schemePair = null;
+ for (Pair pair : fieldKeyPairs) {
+ if (EnrField.ID.equals(pair.getValue0())) {
+ schemePair = pair;
+ break;
+ }
+ }
+ if (schemePair == null) {
+ throw new RuntimeException("ENR scheme (ID) is not defined in key-value pairs");
+ }
+
+ IdentitySchemaInterpreter identitySchemaInterpreter = interpreters.get(schemePair.getValue1());
+ if (identitySchemaInterpreter == null) {
+ throw new RuntimeException(
+ String.format(
+ "No ethereum record interpreter found for identity scheme %s",
+ schemePair.getValue1()));
+ }
+
+ return NodeRecord.fromValues(identitySchemaInterpreter, seq, fieldKeyPairs);
+ }
+
+ public NodeRecord fromBase64(String enrBase64) {
+ return fromBytes(Base64.getUrlDecoder().decode(enrBase64));
+ }
+
+ public NodeRecord fromBytes(BytesValue bytes) {
+ return fromBytes(bytes.extractArray());
+ }
+
+ public NodeRecord fromRlpList(RlpList rlpList) {
+ List values = rlpList.getValues();
+ if (values.size() < 4) {
+ throw new RuntimeException(
+ String.format("Unable to deserialize ENR with less than 4 fields, [%s]", values));
+ }
+
+ // TODO: repair as id is not first now
+ IdentitySchema nodeIdentity = null;
+ boolean idFound = false;
+ for (int i = 2; i < values.size(); i += 2) {
+ RlpString id = (RlpString) values.get(i);
+ if (!"id".equals(new String(id.getBytes()))) {
+ continue;
+ }
+
+ RlpString idVersion = (RlpString) values.get(i + 1);
+ nodeIdentity = IdentitySchema.fromString(new String(idVersion.getBytes()));
+ if (nodeIdentity == null) { // no interpreter for such id
+ throw new RuntimeException(
+ String.format(
+ "Unknown node identity scheme '%s', couldn't create node record.",
+ idVersion.asString()));
+ }
+ idFound = true;
+ break;
+ }
+ if (!idFound) { // no `id` key-values
+ throw new RuntimeException("Unknown node identity scheme, not defined in record ");
+ }
+
+ IdentitySchemaInterpreter identitySchemaInterpreter = interpreters.get(nodeIdentity);
+ if (identitySchemaInterpreter == null) {
+ throw new RuntimeException(
+ String.format(
+ "No Ethereum record interpreter found for identity scheme %s", nodeIdentity));
+ }
+
+ return NodeRecord.fromRawFields(
+ identitySchemaInterpreter,
+ UInt64.fromBytesBigEndian(
+ Bytes8.leftPad(BytesValue.wrap(((RlpString) values.get(1)).getBytes()))),
+ BytesValue.wrap(((RlpString) values.get(0)).getBytes()),
+ values.subList(2, values.size()));
+ }
+
+ public NodeRecord fromBytes(byte[] bytes) {
+ // record = [signature, seq, k, v, ...]
+ RlpList rlpList = (RlpList) RlpDecoder.decode(bytes).getValues().get(0);
+ return fromRlpList(rlpList);
+ }
+}
diff --git a/discovery/src/main/java/org/ethereum/beacon/discovery/message/DiscoveryMessage.java b/discovery/src/main/java/org/ethereum/beacon/discovery/message/DiscoveryMessage.java
new file mode 100644
index 000000000..b68e278ed
--- /dev/null
+++ b/discovery/src/main/java/org/ethereum/beacon/discovery/message/DiscoveryMessage.java
@@ -0,0 +1,11 @@
+package org.ethereum.beacon.discovery.message;
+
+import org.ethereum.beacon.discovery.Protocol;
+import tech.pegasys.artemis.util.bytes.BytesValue;
+
+/** Discovery message */
+public interface DiscoveryMessage {
+ Protocol getProtocol();
+
+ BytesValue getBytes();
+}
diff --git a/discovery/src/main/java/org/ethereum/beacon/discovery/message/DiscoveryV5Message.java b/discovery/src/main/java/org/ethereum/beacon/discovery/message/DiscoveryV5Message.java
new file mode 100644
index 000000000..3889bc37c
--- /dev/null
+++ b/discovery/src/main/java/org/ethereum/beacon/discovery/message/DiscoveryV5Message.java
@@ -0,0 +1,86 @@
+package org.ethereum.beacon.discovery.message;
+
+import org.ethereum.beacon.discovery.Protocol;
+import org.ethereum.beacon.discovery.enr.NodeRecordFactory;
+import org.web3j.rlp.RlpDecoder;
+import org.web3j.rlp.RlpList;
+import org.web3j.rlp.RlpString;
+import org.web3j.rlp.RlpType;
+import tech.pegasys.artemis.util.bytes.BytesValue;
+
+import java.util.List;
+
+public class DiscoveryV5Message implements DiscoveryMessage {
+ private final BytesValue bytes;
+ private List payload = null;
+
+ public DiscoveryV5Message(BytesValue bytes) {
+ this.bytes = bytes;
+ }
+
+ public static DiscoveryV5Message from(V5Message v5Message) {
+ return new DiscoveryV5Message(v5Message.getBytes());
+ }
+
+ @Override
+ public BytesValue getBytes() {
+ return bytes;
+ }
+
+ @Override
+ public Protocol getProtocol() {
+ return Protocol.V5;
+ }
+
+ public MessageCode getCode() {
+ return MessageCode.fromNumber(getBytes().get(0));
+ }
+
+ private synchronized void decode() {
+ if (payload != null) {
+ return;
+ }
+ this.payload =
+ ((RlpList) RlpDecoder.decode(getBytes().slice(1).extractArray()).getValues().get(0))
+ .getValues();
+ }
+
+ public BytesValue getRequestId() {
+ decode();
+ return BytesValue.wrap(((RlpString) payload.get(0)).getBytes());
+ }
+
+ public V5Message create(NodeRecordFactory nodeRecordFactory) {
+ decode();
+ MessageCode code = MessageCode.fromNumber(getBytes().get(0));
+ switch (code) {
+ case PING:
+ {
+ return PingMessage.fromRlp(payload);
+ }
+ case PONG:
+ {
+ return PongMessage.fromRlp(payload);
+ }
+ case FINDNODE:
+ {
+ return FindNodeMessage.fromRlp(payload);
+ }
+ case NODES:
+ {
+ return NodesMessage.fromRlp(payload, nodeRecordFactory);
+ }
+ default:
+ {
+ throw new RuntimeException(
+ String.format(
+ "Creation of discovery V5 messages from code %s is not supported", code));
+ }
+ }
+ }
+
+ @Override
+ public String toString() {
+ return "DiscoveryV5Message{" + "code=" + getCode() + ", bytes=" + getBytes() + '}';
+ }
+}
diff --git a/discovery/src/main/java/org/ethereum/beacon/discovery/message/FindNodeMessage.java b/discovery/src/main/java/org/ethereum/beacon/discovery/message/FindNodeMessage.java
new file mode 100644
index 000000000..7c36db08f
--- /dev/null
+++ b/discovery/src/main/java/org/ethereum/beacon/discovery/message/FindNodeMessage.java
@@ -0,0 +1,72 @@
+package org.ethereum.beacon.discovery.message;
+
+import com.google.common.base.Objects;
+import org.web3j.rlp.RlpEncoder;
+import org.web3j.rlp.RlpList;
+import org.web3j.rlp.RlpString;
+import org.web3j.rlp.RlpType;
+import tech.pegasys.artemis.util.bytes.Bytes1;
+import tech.pegasys.artemis.util.bytes.BytesValue;
+
+import java.util.List;
+
+/**
+ * FINDNODE queries for nodes at the given logarithmic distance from the recipient's node ID. The
+ * node IDs of all nodes in the response must have a shared prefix length of distance with the
+ * recipient's node ID. A request with distance 0 should return the recipient's current record as
+ * the only result.
+ */
+public class FindNodeMessage implements V5Message {
+ // Unique request id
+ private final BytesValue requestId;
+ // The requested log2 distance, a positive integer
+ private final Integer distance;
+
+ public FindNodeMessage(BytesValue requestId, Integer distance) {
+ this.requestId = requestId;
+ this.distance = distance;
+ }
+
+ public static FindNodeMessage fromRlp(List rlpList) {
+ return new FindNodeMessage(
+ BytesValue.wrap(((RlpString) rlpList.get(0)).getBytes()),
+ ((RlpString) rlpList.get(1)).asPositiveBigInteger().intValueExact());
+ }
+
+ @Override
+ public BytesValue getRequestId() {
+ return requestId;
+ }
+
+ public Integer getDistance() {
+ return distance;
+ }
+
+ @Override
+ public BytesValue getBytes() {
+ return Bytes1.intToBytes1(MessageCode.FINDNODE.byteCode())
+ .concat(
+ BytesValue.wrap(
+ RlpEncoder.encode(
+ new RlpList(
+ RlpString.create(requestId.extractArray()), RlpString.create(distance)))));
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) return true;
+ if (o == null || getClass() != o.getClass()) return false;
+ FindNodeMessage that = (FindNodeMessage) o;
+ return Objects.equal(requestId, that.requestId) && Objects.equal(distance, that.distance);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hashCode(requestId, distance);
+ }
+
+ @Override
+ public String toString() {
+ return "FindNodeMessage{" + "requestId=" + requestId + ", distance=" + distance + '}';
+ }
+}
diff --git a/discovery/src/main/java/org/ethereum/beacon/discovery/message/MessageCode.java b/discovery/src/main/java/org/ethereum/beacon/discovery/message/MessageCode.java
new file mode 100644
index 000000000..6c06118b2
--- /dev/null
+++ b/discovery/src/main/java/org/ethereum/beacon/discovery/message/MessageCode.java
@@ -0,0 +1,75 @@
+package org.ethereum.beacon.discovery.message;
+
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * Discovery protocol message types as described in https://github.com/ethereum/devp2p/blob/master/discv5/discv5-wire.md#protocol-messages
+ */
+public enum MessageCode {
+
+ /**
+ * PING checks whether the recipient is alive and informs it about the sender's ENR sequence
+ * number.
+ */
+ PING(0x01),
+
+ /** PONG is the reply to PING. */
+ PONG(0x02),
+
+ /** FINDNODE queries for nodes at the given logarithmic distance from the recipient's node ID. */
+ FINDNODE(0x03),
+
+ /**
+ * NODES is the response to a FINDNODE or TOPICQUERY message. Multiple NODES messages may be sent
+ * as responses to a single query.
+ */
+ NODES(0x04),
+
+ /** Request for {@link #TICKET} by topic. */
+ REQTICKET(0x05),
+
+ /**
+ * TICKET is the response to REQTICKET. It contains a ticket which can be used to register for the
+ * requested topic.
+ */
+ TICKET(0x06),
+
+ /**
+ * REGTOPIC registers the sender for the given topic with a ticket. The ticket must be valid and
+ * its waiting time must have elapsed before using the ticket.
+ */
+ REGTOPIC(0x07),
+
+ /** REGCONFIRMATION is the response to REGTOPIC. */
+ REGCONFIRMATION(0x08),
+
+ /**
+ * TOPICQUERY requests nodes in the topic queue of the given topic. The response is a NODES
+ * message containing node records registered for the topic.
+ */
+ TOPICQUERY(0x09);
+
+ private static final Map codeMap = new HashMap<>();
+
+ static {
+ for (MessageCode type : MessageCode.values()) {
+ codeMap.put(type.code, type);
+ }
+ }
+
+ private int code;
+
+ MessageCode(int code) {
+ this.code = code;
+ }
+
+ public static MessageCode fromNumber(int i) {
+ return codeMap.get(i);
+ }
+
+ public byte byteCode() {
+ return (byte) code;
+ }
+}
diff --git a/discovery/src/main/java/org/ethereum/beacon/discovery/message/NodesMessage.java b/discovery/src/main/java/org/ethereum/beacon/discovery/message/NodesMessage.java
new file mode 100644
index 000000000..faa003767
--- /dev/null
+++ b/discovery/src/main/java/org/ethereum/beacon/discovery/message/NodesMessage.java
@@ -0,0 +1,116 @@
+package org.ethereum.beacon.discovery.message;
+
+import com.google.common.base.Objects;
+import org.ethereum.beacon.discovery.enr.NodeRecord;
+import org.ethereum.beacon.discovery.enr.NodeRecordFactory;
+import org.web3j.rlp.RlpEncoder;
+import org.web3j.rlp.RlpList;
+import org.web3j.rlp.RlpString;
+import org.web3j.rlp.RlpType;
+import tech.pegasys.artemis.util.bytes.Bytes1;
+import tech.pegasys.artemis.util.bytes.BytesValue;
+
+import java.util.List;
+import java.util.function.Supplier;
+import java.util.stream.Collectors;
+
+/**
+ * NODES is the response to a FINDNODE or TOPICQUERY message. Multiple NODES messages may be sent as
+ * responses to a single query.
+ */
+public class NodesMessage implements V5Message {
+ // Unique request id
+ private final BytesValue requestId;
+ // Total number of responses to the request
+ private final Integer total;
+ // List of nodes upon request
+ private final Supplier> nodeRecordsSupplier;
+ // Size of nodes in current response
+ private final Integer nodeRecordsSize;
+ private List nodeRecords = null;
+
+ public NodesMessage(
+ BytesValue requestId,
+ Integer total,
+ Supplier> nodeRecordsSupplier,
+ Integer nodeRecordsSize) {
+ this.requestId = requestId;
+ this.total = total;
+ this.nodeRecordsSupplier = nodeRecordsSupplier;
+ this.nodeRecordsSize = nodeRecordsSize;
+ }
+
+ public static NodesMessage fromRlp(List rlpList, NodeRecordFactory nodeRecordFactory) {
+ RlpList nodeRecords = (RlpList) rlpList.get(2);
+ return new NodesMessage(
+ BytesValue.wrap(((RlpString) rlpList.get(0)).getBytes()),
+ ((RlpString) rlpList.get(1)).asPositiveBigInteger().intValueExact(),
+ () ->
+ nodeRecords.getValues().stream()
+ .map(rl -> nodeRecordFactory.fromRlpList((RlpList) rl))
+ .collect(Collectors.toList()),
+ nodeRecords.getValues().size());
+ }
+
+ @Override
+ public BytesValue getRequestId() {
+ return requestId;
+ }
+
+ public Integer getTotal() {
+ return total;
+ }
+
+ public synchronized List getNodeRecords() {
+ if (nodeRecords == null) {
+ this.nodeRecords = nodeRecordsSupplier.get();
+ }
+ return nodeRecords;
+ }
+
+ public Integer getNodeRecordsSize() {
+ return nodeRecordsSize;
+ }
+
+ @Override
+ public BytesValue getBytes() {
+ return Bytes1.intToBytes1(MessageCode.NODES.byteCode())
+ .concat(
+ BytesValue.wrap(
+ RlpEncoder.encode(
+ new RlpList(
+ RlpString.create(requestId.extractArray()),
+ RlpString.create(total),
+ new RlpList(
+ getNodeRecords().stream()
+ .map(NodeRecord::asRlp)
+ .collect(Collectors.toList()))))));
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) return true;
+ if (o == null || getClass() != o.getClass()) return false;
+ NodesMessage that = (NodesMessage) o;
+ return Objects.equal(requestId, that.requestId)
+ && Objects.equal(total, that.total)
+ && Objects.equal(nodeRecordsSize, that.nodeRecordsSize);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hashCode(requestId, total, nodeRecordsSize);
+ }
+
+ @Override
+ public String toString() {
+ return "NodesMessage{"
+ + "requestId="
+ + requestId
+ + ", total="
+ + total
+ + ", nodeRecordsSize="
+ + nodeRecordsSize
+ + '}';
+ }
+}
diff --git a/discovery/src/main/java/org/ethereum/beacon/discovery/message/PingMessage.java b/discovery/src/main/java/org/ethereum/beacon/discovery/message/PingMessage.java
new file mode 100644
index 000000000..13638b3d0
--- /dev/null
+++ b/discovery/src/main/java/org/ethereum/beacon/discovery/message/PingMessage.java
@@ -0,0 +1,73 @@
+package org.ethereum.beacon.discovery.message;
+
+import com.google.common.base.Objects;
+import org.web3j.rlp.RlpEncoder;
+import org.web3j.rlp.RlpList;
+import org.web3j.rlp.RlpString;
+import org.web3j.rlp.RlpType;
+import tech.pegasys.artemis.util.bytes.Bytes1;
+import tech.pegasys.artemis.util.bytes.Bytes8;
+import tech.pegasys.artemis.util.bytes.BytesValue;
+import tech.pegasys.artemis.util.uint.UInt64;
+
+import java.util.List;
+
+/**
+ * PING checks whether the recipient is alive and informs it about the sender's ENR sequence number.
+ */
+public class PingMessage implements V5Message {
+ // Unique request id
+ private final BytesValue requestId;
+ // Local ENR sequence number of sender
+ private final UInt64 enrSeq;
+
+ public PingMessage(BytesValue requestId, UInt64 enrSeq) {
+ this.requestId = requestId;
+ this.enrSeq = enrSeq;
+ }
+
+ public static PingMessage fromRlp(List rlpList) {
+ return new PingMessage(
+ BytesValue.wrap(((RlpString) rlpList.get(0)).getBytes()),
+ UInt64.fromBytesBigEndian(
+ Bytes8.leftPad(BytesValue.wrap(((RlpString) rlpList.get(1)).getBytes()))));
+ }
+
+ @Override
+ public BytesValue getRequestId() {
+ return requestId;
+ }
+
+ public UInt64 getEnrSeq() {
+ return enrSeq;
+ }
+
+ @Override
+ public BytesValue getBytes() {
+ return Bytes1.intToBytes1(MessageCode.PING.byteCode())
+ .concat(
+ BytesValue.wrap(
+ RlpEncoder.encode(
+ new RlpList(
+ RlpString.create(requestId.extractArray()),
+ RlpString.create(enrSeq.toBI())))));
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) return true;
+ if (o == null || getClass() != o.getClass()) return false;
+ PingMessage that = (PingMessage) o;
+ return Objects.equal(requestId, that.requestId) && Objects.equal(enrSeq, that.enrSeq);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hashCode(requestId, enrSeq);
+ }
+
+ @Override
+ public String toString() {
+ return "PingMessage{" + "requestId=" + requestId + ", enrSeq=" + enrSeq + '}';
+ }
+}
diff --git a/discovery/src/main/java/org/ethereum/beacon/discovery/message/PongMessage.java b/discovery/src/main/java/org/ethereum/beacon/discovery/message/PongMessage.java
new file mode 100644
index 000000000..33c3cf4a7
--- /dev/null
+++ b/discovery/src/main/java/org/ethereum/beacon/discovery/message/PongMessage.java
@@ -0,0 +1,102 @@
+package org.ethereum.beacon.discovery.message;
+
+import com.google.common.base.Objects;
+import org.web3j.rlp.RlpEncoder;
+import org.web3j.rlp.RlpList;
+import org.web3j.rlp.RlpString;
+import org.web3j.rlp.RlpType;
+import tech.pegasys.artemis.util.bytes.Bytes1;
+import tech.pegasys.artemis.util.bytes.Bytes8;
+import tech.pegasys.artemis.util.bytes.BytesValue;
+import tech.pegasys.artemis.util.uint.UInt64;
+
+import java.util.List;
+
+/** PONG is the reply to PING {@link PingMessage} */
+public class PongMessage implements V5Message {
+ // Unique request id
+ private final BytesValue requestId;
+ // Local ENR sequence number of sender
+ private final UInt64 enrSeq;
+ // 16 or 4 byte IP address of the intended recipient
+ private final BytesValue recipientIp;
+ // recipient UDP port, a 16-bit integer
+ private final Integer recipientPort;
+
+ public PongMessage(
+ BytesValue requestId, UInt64 enrSeq, BytesValue recipientIp, Integer recipientPort) {
+ this.requestId = requestId;
+ this.enrSeq = enrSeq;
+ this.recipientIp = recipientIp;
+ this.recipientPort = recipientPort;
+ }
+
+ public static PongMessage fromRlp(List rlpList) {
+ return new PongMessage(
+ BytesValue.wrap(((RlpString) rlpList.get(0)).getBytes()),
+ UInt64.fromBytesBigEndian(
+ Bytes8.leftPad(BytesValue.wrap(((RlpString) rlpList.get(1)).getBytes()))),
+ BytesValue.wrap(((RlpString) rlpList.get(2)).getBytes()),
+ ((RlpString) rlpList.get(3)).asPositiveBigInteger().intValueExact());
+ }
+
+ @Override
+ public BytesValue getRequestId() {
+ return requestId;
+ }
+
+ public UInt64 getEnrSeq() {
+ return enrSeq;
+ }
+
+ public BytesValue getRecipientIp() {
+ return recipientIp;
+ }
+
+ public Integer getRecipientPort() {
+ return recipientPort;
+ }
+
+ @Override
+ public BytesValue getBytes() {
+ return Bytes1.intToBytes1(MessageCode.PONG.byteCode())
+ .concat(
+ BytesValue.wrap(
+ RlpEncoder.encode(
+ new RlpList(
+ RlpString.create(requestId.extractArray()),
+ RlpString.create(enrSeq.toBI()),
+ RlpString.create(recipientIp.extractArray()),
+ RlpString.create(recipientPort)))));
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) return true;
+ if (o == null || getClass() != o.getClass()) return false;
+ PongMessage that = (PongMessage) o;
+ return Objects.equal(requestId, that.requestId)
+ && Objects.equal(enrSeq, that.enrSeq)
+ && Objects.equal(recipientIp, that.recipientIp)
+ && Objects.equal(recipientPort, that.recipientPort);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hashCode(requestId, enrSeq, recipientIp, recipientPort);
+ }
+
+ @Override
+ public String toString() {
+ return "PongMessage{"
+ + "requestId="
+ + requestId
+ + ", enrSeq="
+ + enrSeq
+ + ", recipientIp="
+ + recipientIp
+ + ", recipientPort="
+ + recipientPort
+ + '}';
+ }
+}
diff --git a/discovery/src/main/java/org/ethereum/beacon/discovery/message/V5Message.java b/discovery/src/main/java/org/ethereum/beacon/discovery/message/V5Message.java
new file mode 100644
index 000000000..9fc7af86d
--- /dev/null
+++ b/discovery/src/main/java/org/ethereum/beacon/discovery/message/V5Message.java
@@ -0,0 +1,10 @@
+package org.ethereum.beacon.discovery.message;
+
+import tech.pegasys.artemis.util.bytes.BytesValue;
+
+/** Message of V5 discovery protocol version */
+public interface V5Message {
+ BytesValue getRequestId();
+
+ BytesValue getBytes();
+}
diff --git a/discovery/src/main/java/org/ethereum/beacon/discovery/message/handler/FindNodeHandler.java b/discovery/src/main/java/org/ethereum/beacon/discovery/message/handler/FindNodeHandler.java
new file mode 100644
index 000000000..d5878e27f
--- /dev/null
+++ b/discovery/src/main/java/org/ethereum/beacon/discovery/message/handler/FindNodeHandler.java
@@ -0,0 +1,78 @@
+package org.ethereum.beacon.discovery.message.handler;
+
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+import org.ethereum.beacon.discovery.NodeRecordInfo;
+import org.ethereum.beacon.discovery.NodeSession;
+import org.ethereum.beacon.discovery.enr.NodeRecord;
+import org.ethereum.beacon.discovery.message.DiscoveryV5Message;
+import org.ethereum.beacon.discovery.message.FindNodeMessage;
+import org.ethereum.beacon.discovery.message.NodesMessage;
+import org.ethereum.beacon.discovery.packet.MessagePacket;
+import org.ethereum.beacon.discovery.storage.NodeBucket;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Optional;
+
+public class FindNodeHandler implements MessageHandler {
+ private static final Logger logger = LogManager.getLogger(FindNodeHandler.class);
+ /**
+ * The maximum size of any packet is 1280 bytes. Implementations should not generate or process
+ * packets larger than this size. As per specification the maximum size of an ENR is 300 bytes. A
+ * NODES message containing all FINDNODE response records would be at least 4800 bytes, not
+ * including additional data such as the header. To stay below the size limit, NODES responses are
+ * sent as multiple messages and specify the total number of responses in the message. 4х300 =
+ * 1200 and we always have 80 bytes for everything else.
+ */
+ private static final int MAX_NODES_PER_MESSAGE = 4;
+
+ public FindNodeHandler() {}
+
+ @Override
+ public void handle(FindNodeMessage message, NodeSession session) {
+ Optional nodeBucketOptional = session.getBucket(message.getDistance());
+ List> nodeRecordsList = new ArrayList<>();
+ int total = 0;
+
+ // Repack to lists of MAX_NODES_PER_MESSAGE size
+ List bucketRecords =
+ nodeBucketOptional.isPresent()
+ ? nodeBucketOptional.get().getNodeRecords()
+ : Collections.emptyList();
+ for (NodeRecordInfo nodeRecordInfo : bucketRecords) {
+ if (total % MAX_NODES_PER_MESSAGE == 0) {
+ nodeRecordsList.add(new ArrayList<>());
+ }
+ List currentList = nodeRecordsList.get(nodeRecordsList.size() - 1);
+ currentList.add(nodeRecordInfo.getNode());
+ ++total;
+ }
+ logger.trace(
+ () ->
+ String.format(
+ "Sending %s nodes in reply to request with distance %s in session %s",
+ bucketRecords.size(), message.getDistance(), session));
+
+ // Send
+ if (nodeRecordsList.isEmpty()) {
+ nodeRecordsList.add(Collections.emptyList());
+ }
+ int finalTotal = total;
+ nodeRecordsList.forEach(
+ recordsList ->
+ session.sendOutgoing(
+ MessagePacket.create(
+ session.getHomeNodeId(),
+ session.getNodeRecord().getNodeId(),
+ session.getAuthTag().get(),
+ session.getInitiatorKey(),
+ DiscoveryV5Message.from(
+ new NodesMessage(
+ message.getRequestId(),
+ finalTotal,
+ () -> recordsList,
+ recordsList.size())))));
+ }
+}
diff --git a/discovery/src/main/java/org/ethereum/beacon/discovery/message/handler/MessageHandler.java b/discovery/src/main/java/org/ethereum/beacon/discovery/message/handler/MessageHandler.java
new file mode 100644
index 000000000..dc25df62d
--- /dev/null
+++ b/discovery/src/main/java/org/ethereum/beacon/discovery/message/handler/MessageHandler.java
@@ -0,0 +1,7 @@
+package org.ethereum.beacon.discovery.message.handler;
+
+import org.ethereum.beacon.discovery.NodeSession;
+
+public interface MessageHandler {
+ void handle(Message message, NodeSession session);
+}
diff --git a/discovery/src/main/java/org/ethereum/beacon/discovery/message/handler/NodesHandler.java b/discovery/src/main/java/org/ethereum/beacon/discovery/message/handler/NodesHandler.java
new file mode 100644
index 000000000..70af0f45c
--- /dev/null
+++ b/discovery/src/main/java/org/ethereum/beacon/discovery/message/handler/NodesHandler.java
@@ -0,0 +1,64 @@
+package org.ethereum.beacon.discovery.message.handler;
+
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+import org.ethereum.beacon.discovery.NodeRecordInfo;
+import org.ethereum.beacon.discovery.NodeSession;
+import org.ethereum.beacon.discovery.message.NodesMessage;
+import org.ethereum.beacon.discovery.pipeline.info.FindNodeRequestInfo;
+import org.ethereum.beacon.discovery.pipeline.info.RequestInfo;
+import org.ethereum.beacon.discovery.task.TaskStatus;
+import org.ethereum.beacon.discovery.task.TaskType;
+
+import java.util.Optional;
+
+public class NodesHandler implements MessageHandler {
+ private static final Logger logger = LogManager.getLogger(FindNodeHandler.class);
+
+ @Override
+ public void handle(NodesMessage message, NodeSession session) {
+ // NODES total count handling
+ Optional requestInfoOpt = session.getRequestId(message.getRequestId());
+ if (!requestInfoOpt.isPresent()) {
+ throw new RuntimeException(
+ String.format(
+ "Request #%s not found in session %s when handling message %s",
+ message.getRequestId(), session, message));
+ }
+ FindNodeRequestInfo requestInfo = (FindNodeRequestInfo) requestInfoOpt.get();
+ int newNodesCount =
+ requestInfo.getRemainingNodes() == null
+ ? message.getTotal() - 1
+ : requestInfo.getRemainingNodes() - 1;
+ if (newNodesCount == 0) {
+ session.clearRequestId(message.getRequestId(), TaskType.FINDNODE);
+ } else {
+ session.updateRequestInfo(
+ message.getRequestId(),
+ new FindNodeRequestInfo(
+ TaskStatus.IN_PROCESS,
+ message.getRequestId(),
+ requestInfo.getFuture(),
+ requestInfo.getDistance(),
+ newNodesCount));
+ }
+
+ // Parse node records
+ logger.trace(
+ () ->
+ String.format(
+ "Received %s node records in session %s. Total buckets expected: %s",
+ message.getNodeRecordsSize(), session, message.getTotal()));
+ message
+ .getNodeRecords()
+ .forEach(
+ nodeRecordV5 -> {
+ nodeRecordV5.verify();
+ NodeRecordInfo nodeRecordInfo = NodeRecordInfo.createDefault(nodeRecordV5);
+ if (!session.getNodeTable().getNode(nodeRecordV5.getNodeId()).isPresent()) {
+ session.getNodeTable().save(nodeRecordInfo);
+ }
+ session.putRecordInBucket(nodeRecordInfo);
+ });
+ }
+}
diff --git a/discovery/src/main/java/org/ethereum/beacon/discovery/message/handler/PingHandler.java b/discovery/src/main/java/org/ethereum/beacon/discovery/message/handler/PingHandler.java
new file mode 100644
index 000000000..80e01e662
--- /dev/null
+++ b/discovery/src/main/java/org/ethereum/beacon/discovery/message/handler/PingHandler.java
@@ -0,0 +1,28 @@
+package org.ethereum.beacon.discovery.message.handler;
+
+import org.ethereum.beacon.discovery.NodeSession;
+import org.ethereum.beacon.discovery.enr.EnrField;
+import org.ethereum.beacon.discovery.message.DiscoveryV5Message;
+import org.ethereum.beacon.discovery.message.PingMessage;
+import org.ethereum.beacon.discovery.message.PongMessage;
+import org.ethereum.beacon.discovery.packet.MessagePacket;
+import tech.pegasys.artemis.util.bytes.Bytes4;
+
+public class PingHandler implements MessageHandler {
+ @Override
+ public void handle(PingMessage message, NodeSession session) {
+ PongMessage responseMessage =
+ new PongMessage(
+ message.getRequestId(),
+ session.getNodeRecord().getSeq(),
+ ((Bytes4) session.getNodeRecord().get(EnrField.IP_V4)),
+ (int) session.getNodeRecord().get(EnrField.UDP_V4));
+ session.sendOutgoing(
+ MessagePacket.create(
+ session.getHomeNodeId(),
+ session.getNodeRecord().getNodeId(),
+ session.getAuthTag().get(),
+ session.getInitiatorKey(),
+ DiscoveryV5Message.from(responseMessage)));
+ }
+}
diff --git a/discovery/src/main/java/org/ethereum/beacon/discovery/message/handler/PongHandler.java b/discovery/src/main/java/org/ethereum/beacon/discovery/message/handler/PongHandler.java
new file mode 100644
index 000000000..a212d39c6
--- /dev/null
+++ b/discovery/src/main/java/org/ethereum/beacon/discovery/message/handler/PongHandler.java
@@ -0,0 +1,12 @@
+package org.ethereum.beacon.discovery.message.handler;
+
+import org.ethereum.beacon.discovery.NodeSession;
+import org.ethereum.beacon.discovery.message.PongMessage;
+import org.ethereum.beacon.discovery.task.TaskType;
+
+public class PongHandler implements MessageHandler {
+ @Override
+ public void handle(PongMessage message, NodeSession session) {
+ session.clearRequestId(message.getRequestId(), TaskType.PING);
+ }
+}
diff --git a/discovery/src/main/java/org/ethereum/beacon/discovery/network/DatagramToBytesValue.java b/discovery/src/main/java/org/ethereum/beacon/discovery/network/DatagramToBytesValue.java
new file mode 100644
index 000000000..4be03ad37
--- /dev/null
+++ b/discovery/src/main/java/org/ethereum/beacon/discovery/network/DatagramToBytesValue.java
@@ -0,0 +1,21 @@
+package org.ethereum.beacon.discovery.network;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.socket.DatagramPacket;
+import io.netty.handler.codec.MessageToMessageDecoder;
+import tech.pegasys.artemis.util.bytes.BytesValue;
+
+import java.util.List;
+
+/** UDP Packet -> BytesValue converter with default Netty interface */
+public class DatagramToBytesValue extends MessageToMessageDecoder {
+ @Override
+ protected void decode(ChannelHandlerContext ctx, DatagramPacket msg, List