diff --git a/silkaj/blocks.py b/silkaj/blocks.py
index d72374ce0ca97b845b6a822bb02f1dc057718247..6aa1c76188a6c64643c1307b8eedbc5dab8bd102 100644
--- a/silkaj/blocks.py
+++ b/silkaj/blocks.py
@@ -23,7 +23,7 @@ from duniterpy.documents import Block
from duniterpy.key.verifying_key import VerifyingKey
from silkaj.constants import BMA_MAX_BLOCKS_CHUNK_SIZE
-from silkaj.network_tools import EndPoint
+from silkaj.network_tools import determine_endpoint
from silkaj.tools import message_exit
@@ -36,7 +36,7 @@ If nothing specified, the whole blockchain gets verified.",
@argument("from_block", default=0, type=INT)
@argument("to_block", default=0, type=INT)
def verify_blocks_signatures(from_block, to_block):
- client = Client(EndPoint().BMA_ENDPOINT)
+ client = Client(determine_endpoint())
to_block = check_passed_blocks_range(client, from_block, to_block)
invalid_blocks_signatures = list()
chunks_from = range(from_block, to_block + 1, BMA_MAX_BLOCKS_CHUNK_SIZE)
diff --git a/silkaj/cli.py b/silkaj/cli.py
index d455cf6409d2b0e9afc6c5f6a5db66b1aa9bf964..b4c633d8395d6077f90993667865b62ff722ce1e 100644
--- a/silkaj/cli.py
+++ b/silkaj/cli.py
@@ -16,6 +16,7 @@
import sys
from click import group, help_option, option, pass_context, version_option
+from duniterpy.api.endpoint import endpoint
from silkaj.auth import generate_auth_file
from silkaj.blocks import verify_blocks_signatures
@@ -39,22 +40,17 @@ from silkaj.wot import id_pubkey_correspondence, received_sent_certifications
@help_option("-h", "--help")
@version_option(SILKAJ_VERSION, "-v", "--version")
@option(
- "--peer",
- "-p",
- help="Default endpoint to reach Ğ1 currency by its official node {}\
+ "--endpoint",
+ "-ep",
+ help=f"Default endpoint to reach Ğ1 currency by its official node {endpoint(G1_DEFAULT_ENDPOINT).host}\
This option allows to specify a custom endpoint as follow: <host>:<port>.\
- In case no port is specified, it defaults to 443.".format(
- ":".join(G1_DEFAULT_ENDPOINT)
- ),
+ In case no port is specified, it defaults to 443.",
)
@option(
"--gtest",
"-gt",
is_flag=True,
- help="Default endpoint to reach ĞTest currency by its official node: {}\
-".format(
- ":".join(G1_TEST_DEFAULT_ENDPOINT)
- ),
+ help=f"Default endpoint to reach ĞTest currency by its official node: {endpoint(G1_TEST_DEFAULT_ENDPOINT).host}",
)
@option(
"--auth-scrypt",
@@ -93,7 +89,7 @@ Do not send the document, but display it instead",
@pass_context
def cli(
ctx,
- peer,
+ endpoint,
gtest,
auth_scrypt,
nrp,
@@ -109,7 +105,7 @@ def cli(
ctx.obj = dict()
ctx.ensure_object(dict)
- ctx.obj["PEER"] = peer
+ ctx.obj["ENDPOINT"] = endpoint
ctx.obj["GTEST"] = gtest
ctx.obj["AUTH_SCRYPT"] = auth_scrypt
ctx.obj["AUTH_SCRYPT_PARAMS"] = nrp
diff --git a/silkaj/commands.py b/silkaj/commands.py
index 3b3a81a0e7bd230639692a263655c19711799343..fbad7379187a892653f8fd2fe1dcf81f66092a13 100644
--- a/silkaj/commands.py
+++ b/silkaj/commands.py
@@ -26,7 +26,7 @@ from tabulate import tabulate
from silkaj.blockchain_tools import HeadBlock
from silkaj.constants import ALL, HOUR
-from silkaj.network_tools import ClientInstance, EndPoint
+from silkaj.network_tools import ClientInstance, determine_endpoint
from silkaj.tools import CurrencySymbol
from silkaj.wot_tools import identity_of
@@ -34,13 +34,13 @@ from silkaj.wot_tools import identity_of
@command("info", help="Display information about currency")
def currency_info():
head_block = HeadBlock().head_block
- ep = EndPoint().ep
+ ep = determine_endpoint()
current_time = from_timestamp(head_block["time"], tz="local")
mediantime = from_timestamp(head_block["medianTime"], tz="local")
print(
"Connected to node:",
- ep["domain"],
- ep["port"],
+ ep.host,
+ ep.port,
"\nCurrent block number:",
head_block["number"],
"\nCurrency name:",
@@ -217,17 +217,17 @@ def argos_info():
currency_symbol = CurrencySymbol().symbol
print(currency_symbol, "|")
print("---")
- ep = EndPoint().ep
- if ep["port"] == "443":
- href = "href=https://%s/" % ep["domain"]
+ ep = determine_endpoint()
+ if ep.port == 443:
+ href = f"href=https://{ep.host}/"
else:
- href = "href=http://{}:{}/".format(ep["domain"], ep["port"])
+ href = f"href=http://{ep.host}:{ep.port}/"
current_time = from_timestamp(head_block["time"], tz="local")
mediantime = from_timestamp(head_block["medianTime"], tz="local")
print(
"Connected to node:",
- ep["domain"],
- ep["port"],
+ ep.host,
+ ep.port,
"|",
href,
"\nCurrent block number:",
diff --git a/silkaj/constants.py b/silkaj/constants.py
index e07ce3e7bc04fd98faec819385aa13083d39e721..cbaee25595cb586d914b94634be8dc14e7f9802b 100644
--- a/silkaj/constants.py
+++ b/silkaj/constants.py
@@ -16,8 +16,8 @@
SILKAJ_VERSION = "0.10.0dev"
G1_SYMBOL = "Ğ1"
GTEST_SYMBOL = "ĞTest"
-G1_DEFAULT_ENDPOINT = "g1.duniter.org", "443"
-G1_TEST_DEFAULT_ENDPOINT = "g1-test.duniter.org", "443"
+G1_DEFAULT_ENDPOINT = "BMAS g1.duniter.org 443"
+G1_TEST_DEFAULT_ENDPOINT = "BMAS g1-test.duniter.org 443"
SUCCESS_EXIT_STATUS = 0
FAILURE_EXIT_STATUS = 1
BMA_MAX_BLOCKS_CHUNK_SIZE = 5000
diff --git a/silkaj/network_tools.py b/silkaj/network_tools.py
index 149e959398388fa8eb8a9e83d9000d80f8508290..56597bd75ef76ba2590fc4eb11b4b61a4d3b005d 100644
--- a/silkaj/network_tools.py
+++ b/silkaj/network_tools.py
@@ -13,11 +13,15 @@
# You should have received a copy of the GNU Affero General Public License
# along with Silkaj. If not, see <https://www.gnu.org/licenses/>.
+import re
+import sys
import urllib
+from duniterpy import constants as du_const
+from duniterpy.api import endpoint as ep
from duniterpy.api.client import Client
-from silkaj.constants import G1_DEFAULT_ENDPOINT, G1_TEST_DEFAULT_ENDPOINT
+from silkaj import constants
def singleton(class_):
@@ -31,40 +35,52 @@ def singleton(class_):
return getinstance
-@singleton
-class EndPoint:
- def __init__(self):
- ep = dict()
- try:
- from click.globals import get_current_context
-
- ctx = get_current_context()
- peer = ctx.obj["PEER"]
- gtest = ctx.obj["GTEST"]
- # To be activated when dropping Python 3.5
- # except (ModuleNotFoundError, RuntimeError):
- except:
- peer, gtest = None, None
- if peer:
- if ":" in peer:
- ep["domain"], ep["port"] = peer.rsplit(":", 1)
- else:
- ep["domain"], ep["port"] = peer, "443"
- else:
- ep["domain"], ep["port"] = (
- G1_TEST_DEFAULT_ENDPOINT if gtest else G1_DEFAULT_ENDPOINT
+def determine_endpoint():
+ """
+ Pass custom endpoint, parse through a regex
+ {host|ipv4|[ipv6]}:{port}{/path}
+ ^(?:(HOST)|(IPV4|[(IPV6)]))(?::(PORT))?(?:/(PATH))?$
+ If gtest flag passed, return default gtest endpoint
+ Else, return g1 default endpoint
+ """
+
+ regex = f"^(?:(?P<host>{du_const.HOST_REGEX})|(?P<ipv4>{du_const.IPV4_REGEX})|\
+(?:\\[(?P<ipv6>{du_const.IPV6_REGEX})\\]))(?::(?P<port>{du_const.PORT_REGEX}))?(?:/(?P<path>{du_const.PATH_REGEX}))?$"
+
+ try:
+ from click.globals import get_current_context
+
+ ctx = get_current_context()
+ endpoint = ctx.obj["ENDPOINT"]
+ gtest = ctx.obj["GTEST"]
+ except (ModuleNotFoundError, RuntimeError):
+ endpoint, gtest = None, None
+
+ if endpoint:
+ m = re.search(re.compile(regex), endpoint)
+ if not m:
+ sys.exit(
+ "Error: Passed endpoint is of wrong format.\n\
+Expected format: {host|ipv4|[ipv6]}:{port}{/path}"
)
- if ep["domain"].startswith("[") and ep["domain"].endswith("]"):
- ep["domain"] = ep["domain"][1:-1]
- self.ep = ep
- api = "BMAS" if ep["port"] == "443" else "BASIC_MERKLED_API"
- self.BMA_ENDPOINT = " ".join([api, ep["domain"], ep["port"]])
+ port = int(m["port"]) if m["port"] else 443
+ host, ipv4 = ep.fix_host_ipv4_mix_up(m["host"], m["ipv4"])
+
+ if port == 443:
+ return ep.SecuredBMAEndpoint(host, ipv4, m["ipv6"], port, m["path"])
+ else:
+ return ep.BMAEndpoint(host, ipv4, m["ipv6"], port)
+
+ elif gtest:
+ return ep.endpoint(constants.G1_TEST_DEFAULT_ENDPOINT)
+ else:
+ return ep.endpoint(constants.G1_DEFAULT_ENDPOINT)
@singleton
class ClientInstance:
def __init__(self):
- self.client = Client(EndPoint().BMA_ENDPOINT)
+ self.client = Client(determine_endpoint())
def send_document(bma_path, document):
diff --git a/tests/test_network_tools.py b/tests/test_network_tools.py
index 2c4947a42e1e8aa5b5c8ea29665f70143738a7bc..51f630943ab16b660d2b9aba385c759f2ddc5343 100644
--- a/tests/test_network_tools.py
+++ b/tests/test_network_tools.py
@@ -16,10 +16,55 @@
import urllib
from unittest.mock import patch
+import pytest
from duniterpy.api import bma
+from duniterpy.api import endpoint as du_ep
-from silkaj import network_tools
+from silkaj import constants, network_tools
from silkaj.membership import generate_membership_document
+from tests import helpers
+
+ipv6 = "2001:0db8:85a3:0000:0000:8a2e:0370:7334"
+
+
+@pytest.mark.parametrize(
+ "endpoint, host, ipv4, ipv6, port, path",
+ [
+ ("127.0.0.1", "", "127.0.0.1", None, 443, None),
+ ("127.0.0.1:80", "", "127.0.0.1", None, 80, None),
+ ("127.0.0.1:443", "", "127.0.0.1", None, 443, None),
+ ("127.0.0.1/path", "", "127.0.0.1", None, 443, "path"),
+ ("127.0.0.1:80/path", "", "127.0.0.1", None, 80, "path"),
+ ("domain.tld:80/path", "domain.tld", None, None, 80, "path"),
+ ("localhost:80/path", "localhost", None, None, 80, "path"),
+ (f"[{ipv6}]", None, None, ipv6, 443, None),
+ (f"[{ipv6}]/path", None, None, ipv6, 443, "path"),
+ (f"[{ipv6}]:80/path", None, None, ipv6, 80, "path"),
+ ],
+)
+def test_determine_endpoint_custom(endpoint, host, ipv4, ipv6, port, path):
+ helpers.define_click_context(endpoint)
+ ep = network_tools.determine_endpoint()
+ assert ep.host == host
+ assert ep.ipv4 == ipv4
+ assert ep.ipv6 == ipv6
+ assert ep.port == port
+ if isinstance(ep, du_ep.SecuredBMAEndpoint):
+ assert ep.path == path
+
+
+@pytest.mark.parametrize(
+ "gtest, endpoint",
+ [
+ (True, constants.G1_TEST_DEFAULT_ENDPOINT),
+ (False, constants.G1_DEFAULT_ENDPOINT),
+ ],
+)
+def test_determine_endpoint(gtest, endpoint):
+ helpers.define_click_context(gtest=gtest)
+ ep = network_tools.determine_endpoint()
+ assert ep == du_ep.endpoint(endpoint)
+
# def test_send_document_success(capsys):
# display = capsys.readouterr().out
diff --git a/tests/test_verify_blocks.py b/tests/test_verify_blocks.py
index 223e49c08f678f4658b12a303e7eaeba2838228e..af48140db616c1cff8937da19f688fe3f7ab1e89 100644
--- a/tests/test_verify_blocks.py
+++ b/tests/test_verify_blocks.py
@@ -32,7 +32,7 @@ from silkaj.constants import (
FAILURE_EXIT_STATUS,
SUCCESS_EXIT_STATUS,
)
-from silkaj.network_tools import EndPoint
+from silkaj.network_tools import determine_endpoint
G1_INVALID_BLOCK_SIG = 15144
HEAD_BLOCK = 48000
@@ -47,7 +47,7 @@ def current(self):
)
def test_check_passed_blocks_range(from_block, to_block, capsys, monkeypatch):
monkeypatch.setattr(bma.blockchain, "current", current)
- client = Client(EndPoint().BMA_ENDPOINT)
+ client = Client(determine_endpoint)
# https://medium.com/python-pandemonium/testing-sys-exit-with-pytest-10c6e5f7726f
with pytest.raises(SystemExit) as pytest_wrapped_e:
check_passed_blocks_range(client, from_block, to_block)
@@ -104,7 +104,7 @@ def test_get_chunk_size(from_block, to_block, chunks_from, chunk_from):
@pytest.mark.parametrize("chunk_size, chunk_from", [(2, 1), (5, 10)])
def test_get_chunks(chunk_size, chunk_from):
- client = Client(EndPoint().BMA_ENDPOINT)
+ client = Client(determine_endpoint())
chunk = get_chunk(client, chunk_size, chunk_from)
assert chunk[0]["number"] + chunk_size - 1 == chunk[-1]["number"]