Skip to content

Commit c3b926d

Browse files
unitdevgccunitdevgcc
committed
Dexote kay server pub
1 parent 55ffbcc commit c3b926d

5 files changed

Lines changed: 86 additions & 13 deletions

File tree

‎server/src/main/java/dev/c0redev/volter/ClusterRuntime.java‎

Lines changed: 80 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -46,9 +46,14 @@ static ClusterRuntime get() {
4646
private volatile boolean running;
4747
private volatile Thread worker;
4848
private volatile String lastStateDigest = "";
49+
private volatile String selfDexotePub = "";
4950

5051
private ClusterRuntime() {}
5152

53+
void setSelfDexotePub(String pubBase64) {
54+
selfDexotePub = pubBase64 == null ? "" : pubBase64.trim();
55+
}
56+
5257
void start(Config cfg) {
5358
this.cfg = cfg;
5459
registerSelf();
@@ -75,7 +80,7 @@ void registerSelf() {
7580
endpoint = "http://" + host.trim() + ":" + c.listenPorts().get(0);
7681
dhtRpc = selfDhtRpc(host, c.dhtRpcListenUdp());
7782
}
78-
nodes.put(c.clusterNodeId(), new ClusterNode(c.clusterNodeId(), endpoint, dhtRpc, System.currentTimeMillis(), true));
83+
nodes.put(c.clusterNodeId(), new ClusterNode(c.clusterNodeId(), endpoint, dhtRpc, selfDexotePub, System.currentTimeMillis(), true));
7984
}
8085

8186
private String resolveSelfHost(Config c) {
@@ -169,6 +174,57 @@ boolean isOwnAddress(InetSocketAddress addr) {
169174
}
170175

171176
Optional<InetSocketAddress> resolveClusterExitDialAddress(String hint) {
177+
return resolveClusterExit(hint).map(t -> t.addr);
178+
}
179+
180+
ф static final class ClusterExitTarget {
181+
final InetSocketAddress addr;
182+
final byte[] dexotePub;
183+
184+
ClusterExitTarget(InetSocketAddress addr, byte[] dexotePub) {
185+
this.addr = addr;
186+
this.dexotePub = dexotePub;
187+
}
188+
}
189+
190+
private Optional<byte[]> dexotePubForNode(String nodeId) {
191+
ClusterNode n = nodes.get(nodeId);
192+
if (n == null || n.dexotePub == null || n.dexotePub.isBlank()) {
193+
return Optional.empty();
194+
}
195+
try {
196+
byte[] dec = java.util.Base64.getDecoder().decode(n.dexotePub.trim());
197+
if (dec.length != 32) {
198+
return Optional.empty();
199+
}
200+
return Optional.of(dec);
201+
} catch (IllegalArgumentException e) {
202+
return Optional.empty();
203+
}
204+
}
205+
206+
private Optional<byte[]> dexotePubForAddress(InetSocketAddress want) {
207+
for (String id : nodes.keySet()) {
208+
ClusterNode node = nodes.get(id);
209+
if (node == null || !node.alive) {
210+
continue;
211+
}
212+
Optional<String> ohp = resolveVolterHttpHostPort(id);
213+
if (ohp.isEmpty()) {
214+
continue;
215+
}
216+
try {
217+
InetSocketAddress known = ClusterTcpExitBridge.parseHostPort(ohp.get());
218+
if (known.getPort() == want.getPort() && known.getAddress().equals(want.getAddress())) {
219+
return dexotePubForNode(id);
220+
}
221+
} catch (Exception ignored) {
222+
}
223+
}
224+
return Optional.empty();
225+
}
226+
227+
Optional<ClusterExitTarget> resolveClusterExit(String hint) {
172228
if (hint == null || hint.isBlank()) {
173229
return Optional.empty();
174230
}
@@ -186,7 +242,12 @@ Optional<InetSocketAddress> resolveClusterExitDialAddress(String hint) {
186242
log.info("cluster exit " + h + " resolved to self: " + addr + ", skip bridge");
187243
return Optional.empty();
188244
}
189-
return Optional.of(addr);
245+
Optional<byte[]> pub = dexotePubForNode(h);
246+
if (pub.isEmpty()) {
247+
log.warning("cluster exit " + h + " has no dexotePub in cluster map, skip bridge");
248+
return Optional.empty();
249+
}
250+
return Optional.of(new ClusterExitTarget(addr, pub.get()));
190251
} catch (Exception e) {
191252
return Optional.empty();
192253
}
@@ -199,7 +260,12 @@ Optional<InetSocketAddress> resolveClusterExitDialAddress(String hint) {
199260
return Optional.empty();
200261
}
201262
if (isAuthorizedClusterExit(h)) {
202-
return Optional.of(addr);
263+
Optional<byte[]> pub = dexotePubForAddress(addr);
264+
if (pub.isEmpty()) {
265+
log.warning("cluster exit " + h + " has no dexotePub in cluster map, skip bridge");
266+
return Optional.empty();
267+
}
268+
return Optional.of(new ClusterExitTarget(addr, pub.get()));
203269
}
204270
} catch (Exception ignored) {
205271
}
@@ -252,6 +318,9 @@ String clusterMapJson() {
252318
if (n.dhtRpc != null && !n.dhtRpc.isBlank()) {
253319
sb.append("\"dhtRpc\":\"").append(json(n.dhtRpc)).append("\",");
254320
}
321+
if (n.dexotePub != null && !n.dexotePub.isBlank()) {
322+
sb.append("\"dexotePub\":\"").append(json(n.dexotePub)).append("\",");
323+
}
255324
sb.append("\"ts\":").append(n.lastSeenMs).append(",");
256325
sb.append("\"alive\":").append(n.alive).append("}");
257326
}
@@ -390,7 +459,7 @@ private void markPeerDown(String endpoint) {
390459
for (Map.Entry<String, ClusterNode> e : nodes.entrySet()) {
391460
if (!normalizeEndpoint(e.getValue().endpoint).equals(want)) continue;
392461
ClusterNode n = e.getValue();
393-
nodes.put(e.getKey(), new ClusterNode(n.nodeId, n.endpoint, n.dhtRpc, System.currentTimeMillis(), false));
462+
nodes.put(e.getKey(), new ClusterNode(n.nodeId, n.endpoint, n.dhtRpc, n.dexotePub, System.currentTimeMillis(), false));
394463
}
395464
}
396465

@@ -470,6 +539,7 @@ private int mergeFromJson(String raw) {
470539
String id = row.getOrDefault("id", "").trim();
471540
String endpoint = row.getOrDefault("endpoint", "").trim();
472541
String dhtRpc = row.getOrDefault("dhtRpc", "").trim();
542+
String dexotePub = row.getOrDefault("dexotePub", "").trim();
473543
if (id.isEmpty()) continue;
474544
if (endpoint.isEmpty()) endpoint = "";
475545
long seen = now;
@@ -480,7 +550,7 @@ private int mergeFromJson(String raw) {
480550
} catch (NumberFormatException ignored) {
481551
}
482552
}
483-
nodes.put(id, new ClusterNode(id, endpoint, dhtRpc, seen, true));
553+
nodes.put(id, new ClusterNode(id, endpoint, dhtRpc, dexotePub, seen, true));
484554
merged++;
485555
}
486556
return merged;
@@ -516,10 +586,12 @@ private static List<Map<String, String>> parseNodes(String json) {
516586
String id = jsonField(c, "id");
517587
String endpoint = jsonField(c, "endpoint");
518588
String dhtRpc = jsonField(c, "dhtRpc");
589+
String dexotePub = jsonField(c, "dexotePub");
519590
String ts = jsonFieldLong(c, "ts");
520591
if (id != null) m.put("id", id);
521592
if (endpoint != null) m.put("endpoint", endpoint);
522593
if (dhtRpc != null) m.put("dhtRpc", dhtRpc);
594+
if (dexotePub != null) m.put("dexotePub", dexotePub);
523595
if (ts != null) m.put("ts", ts);
524596
if (!m.isEmpty()) out.add(m);
525597
}
@@ -607,13 +679,15 @@ private static final class ClusterNode {
607679
final String nodeId;
608680
final String endpoint;
609681
final String dhtRpc;
682+
final String dexotePub;
610683
final long lastSeenMs;
611684
final boolean alive;
612685

613-
ClusterNode(String nodeId, String endpoint, String dhtRpc, long lastSeenMs, boolean alive) {
686+
ClusterNode(String nodeId, String endpoint, String dhtRpc, String dexotePub, long lastSeenMs, boolean alive) {
614687
this.nodeId = nodeId;
615688
this.endpoint = endpoint;
616689
this.dhtRpc = dhtRpc;
690+
this.dexotePub = dexotePub == null ? "" : dexotePub;
617691
this.lastSeenMs = lastSeenMs;
618692
this.alive = alive;
619693
}

‎server/src/main/java/dev/c0redev/volter/ClusterTcpExitBridge.java‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -59,7 +59,6 @@ static boolean maybeBridge(
5959
OutputStream clientOut,
6060
Optional<Protocol.ClientOptions> copts,
6161
Socket clientSocket,
62-
byte[] upstreamPub,
6362
long slot)
6463
throws IOException {
6564
if (cfg == null || c == null || clientIn == null || clientOut == null) {
@@ -79,15 +78,16 @@ static boolean maybeBridge(
7978
pref = pref.trim();
8079
boolean strictExit = !cfg.clusterExitFallbackToDirect();
8180
ClusterRuntime rt = ClusterRuntime.get();
82-
Optional<InetSocketAddress> resolved = rt.resolveClusterExitDialAddress(pref);
81+
Optional<ClusterRuntime.ClusterExitTarget> resolved = rt.resolveClusterExit(pref);
8382
if (resolved.isEmpty()) {
8483
log.warning("cluster exit unresolved or not authorized: " + pref + " (strict=" + strictExit + ")");
8584
if (strictExit) {
8685
throw new IOException("cluster exit unresolved or not authorized: " + pref);
8786
}
8887
return false;
8988
}
90-
InetSocketAddress exitAddr = resolved.get();
89+
InetSocketAddress exitAddr = resolved.get().addr;
90+
byte[] upstreamPub = resolved.get().dexotePub;
9191
log.info("cluster exit resolved: " + exitAddr);
9292
int timeoutMs = Math.max(1_000, cfg.quicTcpConnectTimeoutMs());
9393
int readTimeoutMs = RelayCopy.relayReadTimeoutMs(timeoutMs);

‎server/src/main/java/dev/c0redev/volter/ConnectionHandler.java‎

Lines changed: 1 addition & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1035,9 +1035,8 @@ private void handleTcp(
10351035
} else {
10361036
log.info("handleTcp opts: EMPTY");
10371037
}
1038-
byte[] upstreamPub = DexoteUpstream.upstreamPub();
10391038
long upstreamSlot = Dexote.effectiveSlot(System.currentTimeMillis() / 1000L, 0);
1040-
if (ClusterTcpExitBridge.maybeBridge(cfg, c, in, clientOut, copts, s, upstreamPub, upstreamSlot)) {
1039+
if (ClusterTcpExitBridge.maybeBridge(cfg, c, in, clientOut, copts, s, upstreamSlot)) {
10411040

10421041
return;
10431042
}

‎server/src/main/java/dev/c0redev/volter/Main.java‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -41,6 +41,7 @@ public static void main(String[] args) throws Exception {
4141
Dexote.ReplayCache replayCache = new MemReplayCache();
4242

4343
DexoteUpstream.setNodeDexotePub(dexoteKey.pub());
44+
ClusterRuntime.get().setSelfDexotePub(dexoteKey.pubBase64());
4445
Log.setDebug(cfg.debug());
4546
Log.setQuicTrace(cfg.quicTraceLog());
4647
log = Log.logger(Main.class);

‎server/src/main/java/dev/c0redev/volter/QuicServer.java‎

Lines changed: 1 addition & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -338,9 +338,8 @@ public void channelActive(ChannelHandlerContext ctx) {
338338
in,
339339
out,
340340
(connect, rest, copts2) -> {
341-
byte[] upPub = DexoteUpstream.upstreamPub();
342341
long upSlot = Dexote.effectiveSlot(System.currentTimeMillis() / 1000L, 0);
343-
if (ClusterTcpExitBridge.maybeBridge(cfg, connect, rest, out, copts2, null, upPub, upSlot)) {
342+
if (ClusterTcpExitBridge.maybeBridge(cfg, connect, rest, out, copts2, null, upSlot)) {
344343
return;
345344
}
346345
QuicTcpRelay.run(connect, rest, out, cfg.quicTcpConnectTimeoutMs());

0 commit comments

Comments
 (0)