Fix Extra-participant E2E: live contact send and Tor prekey cache path.

Adds connect-contact HTTP endpoint with cached-bundle support, subprocess contact send via docker cp bundle file, and direct Tor prekey fetch to avoid wedging single-worker uvicorn.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
BigBodyCobain
2026-06-14 17:00:13 -06:00
co-authored by Cursor
parent 52a28967a0
commit 7151563a41
5 changed files with 124 additions and 28 deletions
+1
View File
@@ -862,6 +862,7 @@ _ROUTE_TRANSPORT_POLICY: dict[tuple[str, str], RouteTransportPolicy] = {
("POST", "/api/wormhole/gate/messages/decrypt"): _local_only_route_policy("private_control_only"), ("POST", "/api/wormhole/gate/messages/decrypt"): _local_only_route_policy("private_control_only"),
# ── Wormhole DM (strong) ────────────────────────────────────────── # ── Wormhole DM (strong) ──────────────────────────────────────────
("POST", "/api/wormhole/dm/compose"): _local_only_route_policy("private_control_only"), ("POST", "/api/wormhole/dm/compose"): _local_only_route_policy("private_control_only"),
("POST", "/api/wormhole/dm/connect-contact"): _local_only_route_policy("private_control_only"),
("POST", "/api/wormhole/dm/decrypt"): _local_only_route_policy("private_control_only"), ("POST", "/api/wormhole/dm/decrypt"): _local_only_route_policy("private_control_only"),
("POST", "/api/wormhole/dm/mls-key-package"): _local_only_route_policy("private_control_only"), ("POST", "/api/wormhole/dm/mls-key-package"): _local_only_route_policy("private_control_only"),
("POST", "/api/wormhole/dm/register-key"): _local_only_route_policy("private_control_only"), ("POST", "/api/wormhole/dm/register-key"): _local_only_route_policy("private_control_only"),
+22
View File
@@ -330,6 +330,14 @@ class WormholeDmBootstrapDecryptRequest(BaseModel):
ciphertext: str ciphertext: str
class WormholeDmConnectContactRequest(BaseModel):
lookup_token: str = ""
peer_id: str = ""
note: str = ""
lookup_peer_url: str = ""
cached_prekey_bundle: dict[str, Any] | None = None
class WormholeDmInviteImportRequest(BaseModel): class WormholeDmInviteImportRequest(BaseModel):
invite: dict[str, Any] invite: dict[str, Any]
alias: str = "" alias: str = ""
@@ -1089,6 +1097,20 @@ async def api_wormhole_dm_bootstrap_decrypt(request: Request, body: WormholeDmBo
) )
@router.post("/api/wormhole/dm/connect-contact", dependencies=[Depends(require_local_operator)])
@limiter.limit("30/minute")
async def api_wormhole_dm_connect_contact(request: Request, body: WormholeDmConnectContactRequest):
from services.openclaw_infonet import send_contact_request
return send_contact_request(
lookup_token=str(body.lookup_token or ""),
peer_id=str(body.peer_id or ""),
note=str(body.note or ""),
lookup_peer_url=str(body.lookup_peer_url or ""),
cached_prekey_bundle=dict(body.cached_prekey_bundle or {}) if body.cached_prekey_bundle else None,
)
@router.post("/api/wormhole/dm/sender-token", dependencies=[Depends(require_local_operator)]) @router.post("/api/wormhole/dm/sender-token", dependencies=[Depends(require_local_operator)])
@limiter.limit("60/minute") @limiter.limit("60/minute")
async def api_wormhole_dm_sender_token(request: Request, body: WormholeDmSenderTokenRequest): async def api_wormhole_dm_sender_token(request: Request, body: WormholeDmSenderTokenRequest):
@@ -1361,13 +1361,15 @@ def bootstrap_encrypt_for_peer(
plaintext: str, plaintext: str,
*, *,
lookup_token: str = "", lookup_token: str = "",
fetched_bundle: dict[str, Any] | None = None,
) -> dict[str, Any]: ) -> dict[str, Any]:
token = str(lookup_token or "").strip() token = str(lookup_token or "").strip()
peer = str(peer_id or "").strip() peer = str(peer_id or "").strip()
fetched_bundle = fetch_dm_prekey_bundle( if fetched_bundle is None:
agent_id=peer if not token else "", fetched_bundle = fetch_dm_prekey_bundle(
lookup_token=token, agent_id=peer if not token else "",
) lookup_token=token,
)
if not fetched_bundle.get("ok"): if not fetched_bundle.get("ok"):
detail = str(fetched_bundle.get("detail", "") or "") detail = str(fetched_bundle.get("detail", "") or "")
if "root attestation" in detail.lower(): if "root attestation" in detail.lower():
+14 -6
View File
@@ -592,6 +592,7 @@ def send_contact_request(
peer_id: str = "", peer_id: str = "",
note: str = "", note: str = "",
lookup_peer_url: str = "", lookup_peer_url: str = "",
cached_prekey_bundle: dict[str, Any] | None = None,
) -> dict[str, Any]: ) -> dict[str, Any]:
"""Send a first-contact request using a short address or peer id.""" """Send a first-contact request using a short address or peer id."""
from services.mesh.mesh_wormhole_dead_drop import build_contact_offer from services.mesh.mesh_wormhole_dead_drop import build_contact_offer
@@ -604,11 +605,14 @@ def send_contact_request(
return {"ok": False, "detail": "lookup_token or peer_id required"} return {"ok": False, "detail": "lookup_token or peer_id required"}
preferred_peer = str(lookup_peer_url or "").strip().rstrip("/") preferred_peer = str(lookup_peer_url or "").strip().rstrip("/")
bundle = fetch_dm_prekey_bundle( if cached_prekey_bundle and cached_prekey_bundle.get("ok"):
agent_id=peer if not token else "", bundle = dict(cached_prekey_bundle)
lookup_token=token, else:
lookup_peer_urls=[preferred_peer] if preferred_peer else None, bundle = fetch_dm_prekey_bundle(
) agent_id=peer if not token else "",
lookup_token=token,
lookup_peer_urls=[preferred_peer] if preferred_peer else None,
)
if not bundle.get("ok"): if not bundle.get("ok"):
return bundle return bundle
recipient = str(bundle.get("agent_id") or peer).strip() recipient = str(bundle.get("agent_id") or peer).strip()
@@ -621,7 +625,11 @@ def send_contact_request(
dh_algo=str(identity.get("dh_algo") or "X25519"), dh_algo=str(identity.get("dh_algo") or "X25519"),
geo_hint=str(note or ""), geo_hint=str(note or ""),
) )
encrypted = bootstrap_encrypt_for_peer(recipient, offer, lookup_token=token) encrypted = bootstrap_encrypt_for_peer(
recipient,
offer,
fetched_bundle=bundle,
)
if not encrypted.get("ok"): if not encrypted.get("ok"):
return encrypted return encrypted
+81 -18
View File
@@ -18,6 +18,7 @@ import hmac
import os import os
import subprocess import subprocess
import sys import sys
import tempfile
import textwrap import textwrap
import time import time
import urllib.error import urllib.error
@@ -129,15 +130,26 @@ _EMBED_SIGNED_MAILBOX_HELPERS = textwrap.dedent(
def _docker_json(method: str, path: str, body: dict | None = None, *, admin_key: str = "", timeout_s: int = 30) -> dict: def _docker_json(method: str, path: str, body: dict | None = None, *, admin_key: str = "", timeout_s: int = 30) -> dict:
payload = ["docker", "exec", "shadowbroker-backend", "curl", "-s", "--max-time", str(timeout_s)] use_stdin = body is not None
payload = ["docker", "exec"]
if use_stdin:
payload.append("-i")
payload.extend(["shadowbroker-backend", "curl", "-s", "--max-time", str(timeout_s)])
if admin_key: if admin_key:
payload.extend(["-H", f"X-Admin-Key: {admin_key}"]) payload.extend(["-H", f"X-Admin-Key: {admin_key}"])
if body is not None: if body is not None:
payload.extend(["-H", "Content-Type: application/json", "-X", method.upper(), "-d", json.dumps(body)]) payload.extend(["-H", "Content-Type: application/json", "-X", method.upper(), "-d", "@-"])
else: else:
payload.extend(["-X", method.upper()]) payload.extend(["-X", method.upper()])
payload.append(f"http://127.0.0.1:8000{path}") payload.append(f"http://127.0.0.1:8000{path}")
proc = subprocess.run(payload, capture_output=True, text=True, timeout=timeout_s + 15, check=False) proc = subprocess.run(
payload,
input=json.dumps(body) if body is not None else None,
capture_output=True,
text=True,
timeout=timeout_s + 15,
check=False,
)
raw = (proc.stdout or "").strip() raw = (proc.stdout or "").strip()
if not raw: if not raw:
raise RuntimeError(proc.stderr.strip() or f"{method} {path} produced no response") raise RuntimeError(proc.stderr.strip() or f"{method} {path} produced no response")
@@ -330,6 +342,46 @@ def _docker_python(code: str, *, timeout_s: int = 600) -> dict:
return json.loads(line) return json.loads(line)
def _docker_python_contact_send(
*,
handle: str,
peer_id: str,
note: str,
lookup_peer_url: str,
cached_prekey_bundle: dict,
timeout_s: int = 120,
) -> dict:
"""Send contact request in an isolated backend subprocess with a cached prekey bundle."""
with tempfile.NamedTemporaryFile("w", delete=False, suffix=".json", encoding="utf-8") as tmp:
json.dump(cached_prekey_bundle, tmp)
local_path = tmp.name
try:
subprocess.run(
["docker", "cp", local_path, "shadowbroker-backend:/tmp/e2e_prekey_bundle.json"],
check=True,
capture_output=True,
text=True,
)
finally:
os.unlink(local_path)
code = (
"import json\n"
"from services.openclaw_infonet import send_contact_request\n"
'bundle = json.load(open("/tmp/e2e_prekey_bundle.json", encoding="utf-8"))\n'
f"result = send_contact_request(lookup_token={json.dumps(handle)}, "
f"peer_id={json.dumps(peer_id)}, note={json.dumps(note)}, "
f"lookup_peer_url={json.dumps(lookup_peer_url)}, cached_prekey_bundle=bundle)\n"
"print(json.dumps({"
"'ok': bool(result.get('ok')), "
"'send': result, "
"'msg_id': result.get('msg_id',''), "
"'sender_id': result.get('sender_id',''), "
"'recipient_id': result.get('recipient_id','')"
"}))\n"
)
return _docker_python(code, timeout_s=timeout_s)
def _local_compose_cmd(*subcommand: str) -> list[str]: def _local_compose_cmd(*subcommand: str) -> list[str]:
cmd = ["docker", "compose"] cmd = ["docker", "compose"]
for compose_file in LOCAL_COMPOSE_FILES: for compose_file in LOCAL_COMPOSE_FILES:
@@ -2295,6 +2347,15 @@ def _fetch_peer_prekey_bundle(
lookup_token: str = "", lookup_token: str = "",
lookup_peer_url: str = "", lookup_peer_url: str = "",
) -> dict: ) -> dict:
hint = str(lookup_peer_url or (f"http://{PETE_ONION}" if PETE_ONION else "")).strip()
if ".onion" in hint and lookup_token:
try:
bundle = _direct_tor_prekey_bundle(lookup_token, hint)
if bundle.get("ok"):
bundle["source"] = "direct_tor_prekey_bundle"
return bundle
except Exception as exc:
print(json.dumps({"direct_tor_prekey_bundle_fallback": str(exc) or type(exc).__name__}, indent=2))
path = "/api/mesh/dm/prekey-bundle?" path = "/api/mesh/dm/prekey-bundle?"
params: list[str] = [] params: list[str] = []
if lookup_token: if lookup_token:
@@ -2529,8 +2590,8 @@ print(json.dumps({{
return {"ok": False, "detail": str(exc) or type(exc).__name__} return {"ok": False, "detail": str(exc) or type(exc).__name__}
def _direct_tor_prekey_lookup(handle: str, lookup_peer_url: str) -> dict: def _direct_tor_prekey_bundle(handle: str, lookup_peer_url: str) -> dict:
"""Fetch invite-scoped prekey over Tor without blocking local uvicorn on /dm/pubkey.""" """Fetch invite-scoped prekey bundle over Tor without blocking local uvicorn."""
peer = str(lookup_peer_url or "").strip().rstrip("/") peer = str(lookup_peer_url or "").strip().rstrip("/")
if not peer: if not peer:
peer = f"http://{PETE_ONION}".rstrip("/") if PETE_ONION else "" peer = f"http://{PETE_ONION}".rstrip("/") if PETE_ONION else ""
@@ -2560,8 +2621,16 @@ def _direct_tor_prekey_lookup(handle: str, lookup_peer_url: str) -> dict:
) )
raw = (proc.stdout or "").strip() raw = (proc.stdout or "").strip()
if not raw: if not raw:
raise RuntimeError(proc.stderr.strip() or "direct tor prekey lookup produced no response") raise RuntimeError(proc.stderr.strip() or "direct tor prekey bundle produced no response")
payload = json.loads(raw) payload = json.loads(raw)
if not isinstance(payload, dict):
return {"ok": False, "detail": "invalid prekey bundle response"}
return payload
def _direct_tor_prekey_lookup(handle: str, lookup_peer_url: str) -> dict:
"""Fetch invite-scoped pubkey fields over Tor without blocking local uvicorn."""
payload = _direct_tor_prekey_bundle(handle, lookup_peer_url)
if not isinstance(payload, dict) or not payload.get("ok"): if not isinstance(payload, dict) or not payload.get("ok"):
return payload if isinstance(payload, dict) else {"ok": False, "detail": "invalid prekey response"} return payload if isinstance(payload, dict) else {"ok": False, "detail": "invalid prekey response"}
bundle = payload.get("bundle") if isinstance(payload.get("bundle"), dict) else {} bundle = payload.get("bundle") if isinstance(payload.get("bundle"), dict) else {}
@@ -2705,19 +2774,13 @@ def main() -> int:
raise RuntimeError(f"Pete prekey bundle unavailable before contact send: {pete_prekey_bundle}") raise RuntimeError(f"Pete prekey bundle unavailable before contact send: {pete_prekey_bundle}")
print("== step 2: send contact request from local ==") print("== step 2: send contact request from local ==")
send_code = ( send_result = _docker_python_contact_send(
"import json\n" handle=handle,
"from services.openclaw_infonet import send_contact_request\n" peer_id=pete_id,
f"result = send_contact_request(lookup_token={json.dumps(handle)}, note={json.dumps(MARKER)}, lookup_peer_url={json.dumps(lookup_peer_url)})\n" note=MARKER,
"print(json.dumps({" lookup_peer_url=lookup_peer_url,
"'ok': bool(result.get('ok')), " cached_prekey_bundle=pete_prekey_bundle,
"'send': result, "
"'msg_id': result.get('msg_id',''), "
"'sender_id': result.get('sender_id',''), "
"'recipient_id': result.get('recipient_id','')"
"}))\n"
) )
send_result = _docker_python(send_code)
print(json.dumps(send_result, indent=2)) print(json.dumps(send_result, indent=2))
if not send_result.get("ok"): if not send_result.get("ok"):
raise RuntimeError(f"local send failed: {send_result}") raise RuntimeError(f"local send failed: {send_result}")