logos-storage/logos-storage-nim
1
0
Fork 0
mirror of https://github.com/logos-storage/logos-storage-nim.git synced 2026-07-19 15:49:34 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(tools/mix): introduce mix_relay_dht and mix_pool

Browse Source 
Part of https://github.com/logos-storage/logos-storage-pm/issues/13

Signed-off-by: Chrysostomos Nanakos <chris@include.gr>
This commit is contained in:
Chrysostomos Nanakos 2026-06-17 20:47:39 +03:00
parent 2c6cb0161c
commit 0e6b7362e5
No known key found for this signature in database
5 changed files with 1086 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
Makefile
View File

@ -110,6 +110,10 @@ all: | build deps
echo -e $(BUILD_MSG) "build/$@" && \
$(ENV_SCRIPT) nim storage $(NIM_PARAMS) build.nims
mix-tools: | build deps
echo -e $(BUILD_MSG) "build/mix_pool build/mix_relay_dht" && \
$(ENV_SCRIPT) nim mixTools $(NIM_PARAMS) build.nims
# must be included after the default target
-include $(BUILD_SYSTEM_DIR)/makefiles/targets.mk

16
build.nims
View File

@ -66,6 +66,22 @@ task storage, "build logos storage binary":
outname = "storage",
params = "-d:chronicles_runtime_filtering -d:chronicles_log_level=TRACE"
task mixTools, "build mix tools (mix_pool, mix_relay_dht)":
let (desc, ec) = gorgeEx("git describe --always --dirty")
let mixVersion =
if ec == 0 and desc.strip().len > 0: desc.strip() else: "unknown"
let mixParams =
"-d:chronicles_runtime_filtering -d:chronicles_log_level=TRACE " &
"-d:mixVersion:" & mixVersion
buildBinary "mix_pool",
outName = "mix_pool",
srcDir = "tools/mix/",
params = mixParams
buildBinary "mix_relay_dht",
outName = "mix_relay_dht",
srcDir = "tools/mix/",
params = mixParams
task testStorage, "Build & run Logos Storage tests":
test "testStorage", outName = "testStorage"

2
tools/mix/config.nims Normal file
View File

@ -0,0 +1,2 @@
--path:
"../.."

437
tools/mix/mix_pool.nim Normal file
View File

@ -0,0 +1,437 @@
## Copyright (c) 2026 Status Research & Development GmbH
## Licensed under either of
## * Apache License, version 2.0, ([LICENSE-APACHE](LICENSE-APACHE))
## * MIT license ([LICENSE-MIT](LICENSE-MIT))
## at your option.
## This file may not be copied, modified, or distributed except according to
## those terms.
import std/[json, os, parseopt, strformat, strutils]
import pkg/libp2p/crypto/crypto
import pkg/libp2p/crypto/secp
import pkg/libp2p/multiaddress
import pkg/libp2p/peerid
import pkg/libp2p_mix/curve25519
import pkg/libp2p_mix/mix_node
import pkg/stew/byteutils
import pkg/results
const PoolFormatVersion = 1
const MixIdentityFileSize = 2 * FieldElementSize
when not defined(mixVersion):
{.error: "mixVersion must be set at build time via -d:mixVersion:<value>".}
const mixVersion* {.strdefine.} = ""
proc fail(msg: string) {.noreturn.} =
stderr.writeLine msg
quit(1)
proc readBin(path: string): seq[byte] =
if not fileExists(path):
fail "File not found: " & path
try:
cast[seq[byte]](readFile(path))
except IOError as exc:
fail "Failed to read " & path & ": " & exc.msg
proc writeBin(path: string, data: openArray[byte]) =
let parent = parentDir(path)
if parent.len > 0 and not dirExists(parent):
createDir(parent)
try:
writeFile(path, cast[string](@data))
setFilePermissions(path, {fpUserRead, fpUserWrite})
except IOError as exc:
fail "Failed to write " & path & ": " & exc.msg
except OSError as exc:
fail "Failed to set permissions on " & path & ": " & exc.msg
proc pubInfoToJson(info: MixPubInfo): JsonNode =
let (peerId, multiAddr, mixPubKey, libp2pPubKey) = info.get()
%*{
"peerId": $peerId,
"multiAddr": $multiAddr,
"mixPubKey": byteutils.toHex(fieldElementToBytes(mixPubKey)),
"libp2pPubKey": byteutils.toHex(libp2pPubKey.getBytes()),
}
proc pubInfoFromJson(node: JsonNode): MixPubInfo =
let
peerIdStr = node["peerId"].getStr()
multiAddrStr = node["multiAddr"].getStr()
mixPubKeyHex = node["mixPubKey"].getStr()
libp2pPubKeyHex = node["libp2pPubKey"].getStr()
let peerId = PeerId.init(peerIdStr).valueOr:
fail "Invalid peerId in pool entry: " & peerIdStr & " (" & $error & ")"
let multiAddr = MultiAddress.init(multiAddrStr).valueOr:
fail "Invalid multiAddr in pool entry: " & multiAddrStr & " (" & $error & ")"
let mixPubKey = bytesToFieldElement(hexToSeqByte(mixPubKeyHex)).valueOr:
fail "Invalid mixPubKey in pool entry: " & error
let libp2pPubKey = SkPublicKey.init(hexToSeqByte(libp2pPubKeyHex)).valueOr:
fail "Invalid libp2pPubKey in pool entry: " & $error
MixPubInfo.init(peerId, multiAddr, mixPubKey, libp2pPubKey)
proc readPool(path: string): JsonNode =
if not fileExists(path):
return %*{"version": PoolFormatVersion, "relays": newJArray()}
let jsonPool =
try:
readFile(path)
except IOError as exc:
fail "Failed to read pool " & path & ": " & exc.msg
let parsed =
try:
parseJson(jsonPool)
except JsonParsingError as exc:
fail "Pool file is not valid JSON: " & exc.msg
if not parsed.hasKey("version") or parsed["version"].getInt() != PoolFormatVersion:
fail(
"Unsupported pool version (expected " & $PoolFormatVersion & " in " & path & ")"
)
if not parsed.hasKey("relays") or parsed["relays"].kind != JArray:
fail "Pool file missing 'relays' array: " & path
parsed
proc writePool(path: string, pool: JsonNode) =
let parent = parentDir(path)
if parent.len > 0 and not dirExists(parent):
createDir(parent)
try:
writeFile(path, pool.pretty() & "\n")
except IOError as exc:
fail "Failed to write pool " & path & ": " & exc.msg
proc appendOrReplace(pool: JsonNode, entry: JsonNode) =
let
peerId = entry["peerId"].getStr()
relays = pool["relays"]
for i in 0 ..< relays.len:
if relays[i]["peerId"].getStr() == peerId:
relays.elems[i] = entry
return
relays.add(entry)
proc writeMixIdentity(path: string, mixPub, mixPriv: FieldElement) =
let
pubBytes = fieldElementToBytes(mixPub)
privBytes = fieldElementToBytes(mixPriv)
doAssert pubBytes.len == FieldElementSize and privBytes.len == FieldElementSize
writeBin(path, pubBytes & privBytes)
proc readMixIdentity(path: string): tuple[pub: FieldElement, priv: FieldElement] =
let raw = readBin(path)
if raw.len != MixIdentityFileSize:
fail(
"Invalid mix-identity size at " & path & " (expected " & $MixIdentityFileSize &
", got " & $raw.len & ")"
)
let
pub = bytesToFieldElement(raw.toOpenArray(0, FieldElementSize - 1)).valueOr:
fail "Failed to parse mix pub key in " & path & ": " & error
priv = bytesToFieldElement(
raw.toOpenArray(FieldElementSize, MixIdentityFileSize - 1)
).valueOr:
fail "Failed to parse mix priv key in " & path & ": " & error
(pub: pub, priv: priv)
proc writeLibp2pKey(path: string, priv: PrivateKey) =
let bytes = priv.getBytes().valueOr:
fail "Failed to serialize libp2p key: " & $error
writeBin(path, bytes)
proc readLibp2pKey(path: string): PrivateKey =
let bytes = readBin(path)
PrivateKey.init(bytes).valueOr:
fail "Failed to parse libp2p key in " & path & ": " & $error
type
InitArgs = object
pool, outDir, ip: string
count, basePort: int
ExportArgs = object
pool, dataDir, listenIp: string
listenPort: int
ListArgs = object
pool: string
RemoveArgs = object
pool, peerId: string
proc cmdInit(args: InitArgs) =
let rng = newRng()
if rng.isNil:
fail "Failed to create RNG"
var pool = %*{"version": PoolFormatVersion, "relays": newJArray()}
for i in 0 ..< args.count:
let port = args.basePort + i
var nodeInfo = MixNodeInfo.generateRandom(port, rng)
let ma = MultiAddress.init(fmt"/ip4/{args.ip}/tcp/{port}").valueOr:
fail "Failed to construct multiaddr: " & $error
nodeInfo.multiAddr = ma
let libp2pPubProto = PublicKey(scheme: Secp256k1, skkey: nodeInfo.libp2pPubKey)
nodeInfo.peerId = PeerId.init(libp2pPubProto).valueOr:
fail "Failed to derive peerId: " & $error
let nodeDir = args.outDir / fmt"relay_{i}"
writeMixIdentity(nodeDir / "mix-identity", nodeInfo.mixPubKey, nodeInfo.mixPrivKey)
let libp2pPriv = PrivateKey(scheme: Secp256k1, skkey: nodeInfo.libp2pPrivKey)
writeLibp2pKey(nodeDir / "key", libp2pPriv)
pool["relays"].add(pubInfoToJson(nodeInfo.toMixPubInfo()))
writePool(args.pool, pool)
stdout.writeLine "Wrote pool with " & $args.count & " relays to " & args.pool
stdout.writeLine "Per-node identity files under " & args.outDir & "/relay_<i>/"
proc cmdExport(args: ExportArgs) =
if args.listenPort < 1 or args.listenPort > 65535:
fail "--listen-port must be 1..65535"
let
(mixPub, mixPriv) = readMixIdentity(args.dataDir / "mix-identity")
libp2pPriv = readLibp2pKey(args.dataDir / "key")
if libp2pPriv.scheme != Secp256k1:
fail "Mix requires a Secp256k1 libp2p key; got " & $libp2pPriv.scheme
let libp2pPub = libp2pPriv.getPublicKey().valueOr:
fail "Failed to derive libp2p public key: " & $error
let peerId = PeerId.init(libp2pPub).valueOr:
fail "Failed to derive peerId: " & $error
let multiAddr = MultiAddress.init(fmt"/ip4/{args.listenIp}/tcp/{args.listenPort}").valueOr:
fail "Failed to construct multiaddr: " & $error
let
pubInfo = MixPubInfo.init(peerId, multiAddr, mixPub, libp2pPub.skkey)
pool = readPool(args.pool)
pool.appendOrReplace(pubInfoToJson(pubInfo))
writePool(args.pool, pool)
stdout.writeLine "Added/updated relay " & $peerId & " (" & $multiAddr & ") in " &
args.pool
proc cmdList(args: ListArgs) =
let
pool = readPool(args.pool)
relays = pool["relays"]
stdout.writeLine "Pool version " & $pool["version"].getInt() & ", " & $relays.len &
" relays:"
for entry in relays:
stdout.writeLine " " & entry["peerId"].getStr() & " " & entry["multiAddr"].getStr()
proc cmdRemove(args: RemoveArgs) =
let pool = readPool(args.pool)
var
relays = pool["relays"]
filtered = newJArray()
removed = 0
for entry in relays:
if entry["peerId"].getStr() == args.peerId:
inc removed
else:
filtered.add(entry)
pool["relays"] = filtered
writePool(args.pool, pool)
if removed == 0:
stdout.writeLine "No matching peerId in pool; nothing changed."
else:
stdout.writeLine "Removed " & $removed & " entry(ies) for peerId " & args.peerId
proc usage(): string =
"""
mix_pool — manage a Mix relay pool stored as JSON.
Usage:
mix_pool init --pool=<file> --count=N [--ip=<addr>] [--base-port=<n>] [--outdir=<dir>]
mix_pool export --pool=<file> --data-dir=<dir> --listen-ip=<addr> --listen-port=<port>
mix_pool list --pool=<file>
mix_pool remove --pool=<file> --peer-id=<id>
Options (common):
--pool=<file> Path to pool.json (created if absent).
-h, --help Show this help.
-v, --version Show version and revision.
init:
--count=N Number of relays to generate.
--ip=<addr> Public IPv4 set into each relay's multiaddr. (default 127.0.0.1)
--base-port=<n> First TCP port; relay i uses base-port+i. (default 4242)
--outdir=<dir> Where to write each relay's identity files. (default ./relays)
export:
--data-dir=<dir> Existing storage data-dir (contains mix-identity and key).
--listen-ip=<addr> Public IPv4 to embed in the pool entry's multiaddr.
--listen-port=<n> Public TCP port (1..65535).
remove:
--peer-id=<id> Base58 PeerId of the entry to drop.
"""
proc parseSubcommand(): string =
let params = commandLineParams()
if params.len == 0:
stdout.writeLine usage()
quit(1)
let first = params[0]
if first in ["-h", "--help", "help"]:
stdout.writeLine usage()
quit(0)
if first in ["-v", "--version", "version"]:
stdout.writeLine mixVersion
quit(0)
if first.startsWith("-"):
fail "Expected a subcommand as first argument; got: " & first & "\n" & usage()
return first
proc dispatch() =
let sub = parseSubcommand()
var args = commandLineParams()
args.delete(0)
var p = initOptParser(args)
case sub
of "init":
var a =
InitArgs(pool: "", outDir: "./relays", ip: "127.0.0.1", count: 0, basePort: 4242)
while true:
p.next()
case p.kind
of cmdEnd:
break
of cmdShortOption, cmdLongOption:
case p.key
of "help", "h":
stdout.writeLine usage()
quit(0)
of "pool":
a.pool = expandTilde(p.val)
of "count":
try:
a.count = parseInt(p.val)
except ValueError:
fail "init: --count must be an integer, got: " & p.val
of "ip":
a.ip = p.val
of "base-port":
try:
a.basePort = parseInt(p.val)
except ValueError:
fail "init: --base-port must be an integer, got: " & p.val
of "outdir":
a.outDir = expandTilde(p.val)
else:
fail "init: unknown flag --" & p.key
of cmdArgument:
stderr.writeLine usage()
quit(1)
if a.pool.len == 0:
fail "init: --pool=<file> is required"
if a.count < 1:
fail "init: --count=<n> must be >= 1"
cmdInit(a)
of "export":
var a = ExportArgs(pool: "", dataDir: "", listenIp: "", listenPort: 0)
while true:
p.next()
case p.kind
of cmdEnd:
break
of cmdShortOption, cmdLongOption:
case p.key
of "help", "h":
stdout.writeLine usage()
quit(0)
of "pool":
a.pool = expandTilde(p.val)
of "data-dir":
a.dataDir = expandTilde(p.val)
of "listen-ip":
a.listenIp = p.val
of "listen-port":
try:
a.listenPort = parseInt(p.val)
except ValueError:
fail "export: --listen-port must be an integer, got: " & p.val
else:
fail "export: unknown flag --" & p.key
of cmdArgument:
stderr.writeLine usage()
quit(1)
if a.pool.len == 0:
fail "export: --pool=<file> is required"
if a.dataDir.len == 0:
fail "export: --data-dir=<dir> is required"
if a.listenIp.len == 0:
fail "export: --listen-ip=<addr> is required"
if a.listenPort == 0:
fail "export: --listen-port=<port> is required"
cmdExport(a)
of "list":
var a = ListArgs(pool: "")
while true:
p.next()
case p.kind
of cmdEnd:
break
of cmdShortOption, cmdLongOption:
case p.key
of "help", "h":
stdout.writeLine usage()
quit(0)
of "pool":
a.pool = expandTilde(p.val)
else:
fail "list: unknown flag --" & p.key
of cmdArgument:
stderr.writeLine usage()
quit(1)
if a.pool.len == 0:
fail "list: --pool=<file> is required"
cmdList(a)
of "remove":
var a = RemoveArgs(pool: "", peerId: "")
while true:
p.next()
case p.kind
of cmdEnd:
break
of cmdShortOption, cmdLongOption:
case p.key
of "help", "h":
stdout.writeLine usage()
quit(0)
of "pool":
a.pool = expandTilde(p.val)
of "peer-id":
a.peerId = p.val
else:
fail "remove: unknown flag --" & p.key
of cmdArgument:
stderr.writeLine usage()
quit(1)
if a.pool.len == 0:
fail "remove: --pool=<file> is required"
if a.peerId.len == 0:
fail "remove: --peer-id=<id> is required"
cmdRemove(a)
else:
fail "Unknown subcommand: " & sub & "\n" & usage()
when isMainModule:
dispatch()

627
tools/mix/mix_relay_dht.nim Normal file
View File

@ -0,0 +1,627 @@
## Copyright (c) 2026 Status Research & Development GmbH
## Licensed under either of
## * Apache License, version 2.0, ([LICENSE-APACHE](LICENSE-APACHE))
## * MIT license ([LICENSE-MIT](LICENSE-MIT))
## at your option.
## This file may not be copied, modified, or distributed except according to
## those terms.
import std/[net, os, parseopt, strformat, strutils]
import pkg/chronos
import pkg/chronicles
import
pkg/libp2p/
[builders, cid, multiaddress, peerid, routing_record, signed_envelope, switch]
import pkg/libp2p/crypto/crypto
import pkg/libp2p/crypto/secp
import pkg/libp2p/protocols/protocol
import pkg/libp2p/stream/connection
import pkg/libp2p_mix
import pkg/libp2p_mix/[curve25519, mix_node]
import pkg/libp2p/crypto/curve25519 as libp2p_curve25519
import pkg/results
import pkg/codexdht/discv5/[protocol as discv5, routing_table]
from pkg/nimcrypto import keccak256
import pkg/storage/dht_proxy/protocol
when defined(posix):
import std/posix
const MixIdentityFileSize = 2 * FieldElementSize
when not defined(mixVersion):
{.error: "mixVersion must be set at build time via -d:mixVersion:<value>".}
const mixVersion* {.strdefine.} = ""
logScope:
topics = "mix relay dht"
proc fail(msg: string) {.noreturn.} =
stderr.writeLine msg
quit(1)
proc readBin(path: string): seq[byte] =
if not fileExists(path):
fail "File not found: " & path
try:
cast[seq[byte]](readFile(path))
except IOError as exc:
fail "Failed to read " & path & ": " & exc.msg
proc loadMixKeys(path: string): tuple[pub, priv: FieldElement] =
let raw = readBin(path)
if raw.len != MixIdentityFileSize:
fail(
"Invalid mix-identity size at " & path & " (expected " & $MixIdentityFileSize &
", got " & $raw.len & ")"
)
let
pub = bytesToFieldElement(raw.toOpenArray(0, FieldElementSize - 1)).valueOr:
fail "Failed to parse mix pub key in " & path & ": " & error
priv = bytesToFieldElement(
raw.toOpenArray(FieldElementSize, MixIdentityFileSize - 1)
).valueOr:
fail "Failed to parse mix priv key in " & path & ": " & error
if libp2p_curve25519.public(priv) != pub:
fail "Mix identity in " & path & " is inconsistent: pub does not match priv"
(pub: pub, priv: priv)
proc loadLibp2pKey(path: string): PrivateKey =
let bytes = readBin(path)
PrivateKey.init(bytes).valueOr:
fail "Failed to parse libp2p key in " & path & ": " & $error
proc writeBin(path: string, data: openArray[byte]) =
let parent = parentDir(path)
if parent.len > 0 and not dirExists(parent):
createDir(parent)
try:
writeFile(path, cast[string](@data))
setFilePermissions(path, {fpUserRead, fpUserWrite})
except IOError as exc:
fail "Failed to write " & path & ": " & exc.msg
except OSError as exc:
fail "Failed to set permissions on " & path & ": " & exc.msg
proc generateKeys(dataDir: string) =
let
mixIdentityPath = dataDir / "mix-identity"
libp2pKeyPath = dataDir / "key"
if not dirExists(dataDir):
try:
createDir(dataDir)
except OSError as exc:
fail "Failed to create --data-dir " & dataDir & ": " & exc.msg
let rng = newRng()
if rng.isNil:
fail "Failed to create RNG"
let (mixPriv, mixPub) = generateKeyPair().valueOr:
fail "Failed to generate mix keypair: " & error
writeBin(mixIdentityPath, fieldElementToBytes(mixPub) & fieldElementToBytes(mixPriv))
let libp2pPair = SkKeyPair.random(rng)
let libp2pPriv = PrivateKey(scheme: Secp256k1, skkey: libp2pPair.seckey)
let libp2pBytes = libp2pPriv.getBytes().valueOr:
fail "Failed to serialize libp2p key: " & $error
writeBin(libp2pKeyPath, libp2pBytes)
notice "Generated fresh identity",
dataDir = dataDir, mixIdentity = mixIdentityPath, libp2pKey = libp2pKeyPath
proc toNodeId(c: Cid): NodeId =
readUintBE[256](keccak256.digest(c.data.buffer).data)
type DhtProxyProtocol = ref object of LPProtocol
dht: discv5.Protocol
inFlight: int
maxInFlight: int
proc handleFindProviders(
self: DhtProxyProtocol, queryBytes: seq[byte]
): Future[LookupResponse] {.async: (raises: [CancelledError]).} =
let c = Cid.init(queryBytes).valueOr:
warn "Invalid CID in lookup request"
return LookupResponse(status: ResponseStatus.Error, errorKind: ErrorKind.InvalidCid)
let providers =
try:
(await self.dht.getProviders(c.toNodeId())).valueOr:
warn "discv5 getProviders failed", err = $error
return
LookupResponse(status: ResponseStatus.Error, errorKind: ErrorKind.Internal)
except CancelledError as exc:
raise exc
except CatchableError as exc:
warn "discv5 getProviders raised", err = exc.msg
return LookupResponse(status: ResponseStatus.Error, errorKind: ErrorKind.Internal)
if providers.len == 0:
return LookupResponse(status: ResponseStatus.NotFound)
var encoded = newSeqOfCap[seq[byte]](providers.len)
for rec in providers:
let bytes = rec.encode().valueOr:
warn "Failed to encode SignedPeerRecord", err = error
continue
encoded.add(bytes)
if encoded.len == 0:
return LookupResponse(status: ResponseStatus.Error, errorKind: ErrorKind.Internal)
let packed = packProviders(encoded, MaxLookupResponseBytes).valueOr:
return LookupResponse(status: ResponseStatus.Error, errorKind: error)
LookupResponse(status: ResponseStatus.Ok, providers: packed)
proc handleLookupRequest(
self: DhtProxyProtocol, conn: Connection
) {.async: (raises: [CancelledError]).} =
try:
if self.inFlight >= self.maxInFlight:
debug "DHT proxy at capacity, replying TooBusy",
inFlight = self.inFlight, max = self.maxInFlight
await conn.writeLp(
LookupResponse(status: ResponseStatus.Error, errorKind: ErrorKind.TooBusy).encode()
)
return
inc self.inFlight
defer:
dec self.inFlight
let
reqBytes = await conn.readLp(MaxLookupRequestBytes)
req = LookupRequest.decode(reqBytes).valueOr:
warn "Failed to decode lookup request"
await conn.writeLp(
LookupResponse(
status: ResponseStatus.Error, errorKind: ErrorKind.DecodeFailed
).encode()
)
return
let resp =
case req.queryType
of FindProviders:
await self.handleFindProviders(req.queryBytes)
await conn.writeLp(resp.encode())
except CancelledError as exc:
raise exc
except LPStreamError as exc:
warn "Stream error", err = exc.msg
except CatchableError as exc:
warn "Handler error", err = exc.msg
proc new(
T: type DhtProxyProtocol,
dht: discv5.Protocol,
maxInFlight: int = DefaultMaxInFlightLookups,
): DhtProxyProtocol =
let self = DhtProxyProtocol(dht: dht, maxInFlight: maxInFlight)
proc handler(
conn: Connection, proto: string
): Future[void] {.async: (raises: [CancelledError]).} =
try:
await self.handleLookupRequest(conn)
finally:
await noCancel conn.close()
self.handler = handler
self.codec = DhtProxyCodec
self
type Conf = object
dataDir: string
listenIp: string
listenPort: int
discPort: int
bootstrapNodes: seq[SignedPeerRecord]
logLevel: string
logFile: string
generate: bool
noDhtProxy: bool
maxInFlight: int
proc usage(): string =
"""
mix_relay_dht — standalone Mix relay + DHT proxy daemon.
Usage:
mix_relay_dht --data-dir=<dir> --listen-ip=<addr> --listen-port=<port>
--disc-port=<port>
[--bootstrap-node=<spr> ...] [--log-level=<lvl>] [--generate]
Options:
--data-dir=<dir> Directory holding identity files (key + mix-identity).
--listen-ip=<addr> Public IPv4 to bind/announce for libp2p TCP.
--listen-port=<n> libp2p TCP port (Mix relay + DHT proxy share this).
--disc-port=<n> discv5 UDP port.
--bootstrap-node=<spr> Repeatable. SPR of a discv5 bootstrap peer.
--log-level=<lvl> TRACE | DEBUG | INFO | NOTICE | WARN | ERROR | FATAL | NONE
(default: INFO)
--log-file=<path> Write logs to <path> instead of stdout.
--generate Generate fresh identity files if data-dir is empty.
--no-dht-proxy Run as a pure Mix relay.
Conflicts with --disc-port and --bootstrap-node.
--max-inflight=<n> Max concurrent DHT proxy lookups (default: 100).
-h, --help Show this help.
-v, --version Show version and revision.
"""
proc parseSpr(raw: string): SignedPeerRecord =
var spr: SignedPeerRecord
if not spr.fromURI(raw):
fail "Invalid --bootstrap-node SPR: " & raw
spr
proc parseConf(): Conf =
result = Conf(
dataDir: "",
listenIp: "",
listenPort: 0,
discPort: 0,
bootstrapNodes: @[],
logLevel: "INFO",
logFile: "",
generate: false,
noDhtProxy: false,
maxInFlight: DefaultMaxInFlightLookups,
)
var p = initOptParser(commandLineParams())
while true:
p.next()
case p.kind
of cmdEnd:
break
of cmdShortOption, cmdLongOption:
case p.key
of "help", "h":
stdout.writeLine usage()
quit(0)
of "version", "v":
stdout.writeLine mixVersion
quit(0)
of "data-dir":
result.dataDir = expandTilde(p.val)
of "listen-ip":
result.listenIp = p.val
of "listen-port":
try:
result.listenPort = parseInt(p.val)
except ValueError:
fail "--listen-port must be an integer, got: " & p.val
of "disc-port":
try:
result.discPort = parseInt(p.val)
except ValueError:
fail "--disc-port must be an integer, got: " & p.val
of "bootstrap-node":
result.bootstrapNodes.add(parseSpr(p.val))
of "log-level":
try:
discard parseEnum[LogLevel](p.val)
except ValueError:
fail "Invalid --log-level: " & p.val &
" (use TRACE|DEBUG|INFO|NOTICE|WARN|ERROR|FATAL|NONE)"
result.logLevel = p.val
of "log-file":
result.logFile = expandTilde(p.val)
of "generate":
result.generate = true
of "no-dht-proxy":
result.noDhtProxy = true
of "max-inflight":
try:
result.maxInFlight = parseInt(p.val)
except ValueError:
fail "--max-inflight must be an integer, got: " & p.val
if result.maxInFlight < 1:
fail "--max-inflight must be >= 1, got: " & $result.maxInFlight
else:
fail "Unknown flag: --" & p.key
of cmdArgument:
stderr.writeLine usage()
quit(1)
if result.dataDir.len == 0:
fail "--data-dir=<dir> is required"
if result.listenIp.len == 0:
fail "--listen-ip=<addr> is required"
if result.listenPort == 0:
fail "--listen-port=<port> is required"
if result.listenPort < 1 or result.listenPort > 65535:
fail "--listen-port out of range: " & $result.listenPort & " (must be 1..65535)"
if result.noDhtProxy:
if result.discPort != 0:
fail "--no-dht-proxy conflicts with --disc-port"
if result.bootstrapNodes.len > 0:
fail "--no-dht-proxy conflicts with --bootstrap-node"
else:
if result.discPort == 0:
fail "--disc-port=<port> is required"
if result.discPort < 1 or result.discPort > 65535:
fail "--disc-port out of range: " & $result.discPort & " (must be 1..65535)"
var shutdownRequested = false
proc requestShutdown() =
shutdownRequested = true
proc controlCHandler() {.noconv.} =
requestShutdown()
when defined(posix):
proc sigtermHandler(signal: cint) {.noconv.} =
requestShutdown()
proc runRelayOnly(
conf: Conf,
switch: Switch,
mixProto: MixProtocol,
peerId: PeerId,
tcpAddr: MultiAddress,
) {.async: (raises: [CatchableError]).} =
try:
await mixProto.start()
except CatchableError as exc:
raise newException(CatchableError, "MixProtocol start failed: " & exc.msg)
switch.mount(mixProto)
try:
await switch.start()
except CatchableError as exc:
raise newException(CatchableError, "libp2p switch start failed: " & exc.msg)
notice "Mix relay started (no DHT proxy)",
peerId = peerId, tcp = $tcpAddr, dataDir = conf.dataDir
try:
while not shutdownRequested:
await sleepAsync(200.milliseconds)
finally:
notice "Stopping"
await switch.stop()
notice "Stopped"
proc runWithDhtProxy(
conf: Conf,
switch: Switch,
mixProto: MixProtocol,
libp2pPriv: PrivateKey,
peerId: PeerId,
listenIp: IpAddress,
tcpAddr: MultiAddress,
) {.async: (raises: [CatchableError]).} =
let udpAddr = MultiAddress.init(fmt"/ip4/{$listenIp}/udp/{conf.discPort}").valueOr:
raise newException(ValueError, "Invalid discv5 multiaddr: " & $error)
let dhtRecord = SignedPeerRecord.init(libp2pPriv, PeerRecord.init(peerId, @[udpAddr])).valueOr:
raise newException(ValueError, "Failed to build DHT SPR: " & $error)
let discoveryConfig =
DiscoveryConfig(tableIpLimits: DefaultTableIpLimits, bitsPerHop: DefaultBitsPerHop)
let dht = newProtocol(
libp2pPriv,
bindIp = listenIp,
bindPort = Port(conf.discPort),
record = dhtRecord,
bootstrapRecords = conf.bootstrapNodes,
rng = newRng(),
providers =
ProvidersManager.new(SQLiteDatastore.new(Memory).expect("Should not fail")),
config = discoveryConfig,
)
let maxReplyBytes = getMaxMessageSizeForCodec(DhtProxyCodec, 0).valueOr:
raise
newException(ValueError, "DhtProxyCodec does not fit Sphinx payload: " & error)
mixProto.registerDestReadBehavior(DhtProxyCodec, readLp(maxReplyBytes))
let proxyProto = DhtProxyProtocol.new(dht, maxInFlight = conf.maxInFlight)
try:
await mixProto.start()
except CatchableError as exc:
raise newException(CatchableError, "MixProtocol start failed: " & exc.msg)
switch.mount(mixProto)
try:
await proxyProto.start()
except CatchableError as exc:
raise newException(CatchableError, "DhtProxyProtocol start failed: " & exc.msg)
switch.mount(proxyProto)
try:
dht.open()
await dht.start()
except CatchableError as exc:
raise newException(CatchableError, "discv5 start failed: " & exc.msg)
try:
await switch.start()
except CatchableError as exc:
raise newException(CatchableError, "libp2p switch start failed: " & exc.msg)
let mixNodeRecord = SignedPeerRecord.init(
libp2pPriv, PeerRecord.init(peerId, @[tcpAddr])
).valueOr:
raise newException(ValueError, "Failed to build mix node SPR: " & $error)
let
mixNodeSprStr = mixNodeRecord.toURI()
dhtSprStr = dht.localNode.record.toURI()
mixNodeSprPath = conf.dataDir / "mix_node.spr"
dhtSprPath = conf.dataDir / "dht.spr"
try:
writeFile(mixNodeSprPath, mixNodeSprStr)
except IOError as exc:
raise newException(
CatchableError,
"Failed to write mix node SPR file " & mixNodeSprPath & ": " & exc.msg,
)
try:
writeFile(dhtSprPath, dhtSprStr)
except IOError as exc:
raise newException(
CatchableError, "Failed to write DHT SPR file " & dhtSprPath & ": " & exc.msg
)
notice "Mix relay and DHT proxy started",
peerId = peerId, tcp = $tcpAddr, udp = $udpAddr, dataDir = conf.dataDir
notice "DHT bootstrap SPR", spr = dhtSprStr, file = dhtSprPath
notice "Mix node SPR", spr = mixNodeSprStr, file = mixNodeSprPath
try:
while not shutdownRequested:
await sleepAsync(200.milliseconds)
finally:
notice "Stopping"
try:
await noCancel dht.closeWait()
except CatchableError as exc:
warn "discv5 close error", err = exc.msg
await switch.stop()
notice "Stopped"
proc run(conf: Conf) {.async: (raises: [CatchableError]).} =
let
mixIdentityPath = conf.dataDir / "mix-identity"
libp2pKeyPath = conf.dataDir / "key"
mixIdentityExists = fileExists(mixIdentityPath)
libp2pKeyExists = fileExists(libp2pKeyPath)
if not mixIdentityExists and not libp2pKeyExists:
if not conf.generate:
fail(
"No identity files in --data-dir " & conf.dataDir &
". Either provide them or pass --generate to create fresh keys."
)
generateKeys(conf.dataDir)
elif mixIdentityExists xor libp2pKeyExists:
fail(
"Partial identity in --data-dir " & conf.dataDir &
" (one of mix-identity / key is missing). Aborting."
)
elif conf.generate:
warn "Ignoring --generate: identity files already exist in --data-dir",
dataDir = conf.dataDir
let
(mixPub, mixPriv) = loadMixKeys(mixIdentityPath)
libp2pPriv = loadLibp2pKey(libp2pKeyPath)
if libp2pPriv.scheme != Secp256k1:
raise newException(
ValueError, "Mix requires a Secp256k1 libp2p key; got " & $libp2pPriv.scheme
)
let libp2pPub = libp2pPriv.getPublicKey().valueOr:
raise newException(ValueError, "Failed to derive libp2p public key: " & $error)
let peerId = PeerId.init(libp2pPub).valueOr:
raise newException(ValueError, "Failed to derive peerId: " & $error)
let listenIp =
try:
parseIpAddress(conf.listenIp)
except ValueError as exc:
raise newException(ValueError, "Invalid --listen-ip: " & exc.msg)
let tcpAddr = MultiAddress.init(fmt"/ip4/{$listenIp}/tcp/{conf.listenPort}").valueOr:
raise newException(ValueError, "Invalid libp2p multiaddr: " & $error)
let nodeInfo = initMixNodeInfo(
peerId = peerId,
multiAddr = tcpAddr,
mixPubKey = mixPub,
mixPrivKey = mixPriv,
libp2pPubKey = libp2pPub.skkey,
libp2pPrivKey = libp2pPriv.skkey,
)
let switch = SwitchBuilder
.new()
.withPrivateKey(libp2pPriv)
.withAddresses(@[tcpAddr])
.withRng(newRng())
.withNoise()
.withYamux()
.withTcpTransport({ServerFlags.ReuseAddr, ServerFlags.TcpNoDelay})
.build()
let mixProto = MixProtocol.new(nodeInfo, switch)
if conf.noDhtProxy:
await runRelayOnly(conf, switch, mixProto, peerId, tcpAddr)
else:
await runWithDhtProxy(conf, switch, mixProto, libp2pPriv, peerId, listenIp, tcpAddr)
var logFileHandle: File
proc setupLogging(conf: Conf) =
proc writeAndFlush(f: File, msg: LogOutputStr) =
try:
f.write(msg)
f.flushFile()
except IOError as err:
logLoggingFailure(cstring(msg), err)
proc noOutput(logLevel: LogLevel, msg: LogOutputStr) =
discard
proc stdoutWriter(logLevel: LogLevel, msg: LogOutputStr) =
writeAndFlush(stdout, msg)
defaultChroniclesStream.outputs[1].writer = noOutput
if conf.logFile.len == 0:
defaultChroniclesStream.outputs[0].writer = stdoutWriter
defaultChroniclesStream.outputs[2].writer = noOutput
return
try:
logFileHandle = open(conf.logFile, fmWrite)
except IOError as exc:
fail "Failed to open --log-file " & conf.logFile & ": " & exc.msg
proc fileWriter(logLevel: LogLevel, msg: LogOutputStr) =
writeAndFlush(logFileHandle, msg)
defaultChroniclesStream.outputs[0].writer = noOutput
defaultChroniclesStream.outputs[2].writer = fileWriter
proc main() =
let conf = parseConf()
when defined(chronicles_runtime_filtering):
setLogLevel(parseEnum[LogLevel](conf.logLevel))
setupLogging(conf)
try:
setControlCHook(controlCHandler)
except Exception as exc:
warn "Cannot set ctrl-c handler", msg = exc.msg
when defined(posix):
discard posix.signal(SIGTERM, sigtermHandler)
try:
waitFor run(conf)
except CatchableError as exc:
fatal "Mix relay + DHT proxy aborted", err = exc.msg
quit(1)
when isMainModule:
main()