()
| 106 | |
| 107 | |
| 108 | def main(): |
| 109 | ap = argparse.ArgumentParser() |
| 110 | ap.add_argument("--port", type=int, default=14242, |
| 111 | help="UDP port the Python receiver listens on (default 14242 to avoid clashing with the system RNS instance on 4242)") |
| 112 | ap.add_argument("--timeout", type=float, default=60.0, |
| 113 | help="seconds to wait for a resource before giving up") |
| 114 | ap.add_argument("--identity-file", default=None, |
| 115 | help="path to a saved RNS Identity; created if missing") |
| 116 | ap.add_argument("--expect-sha256", default=None, |
| 117 | help="hex sha256 of expected payload; compared on receive") |
| 118 | args = ap.parse_args() |
| 119 | |
| 120 | config_dir = tempfile.mkdtemp(prefix="rns_interop_recv_") |
| 121 | # RNS expects the storage/resources/ subdirectory to exist (used to |
| 122 | # stage incoming resources to disk before the assembled callback). The |
| 123 | # default ~/.reticulum auto-creates it; with a fresh tempdir we have to. |
| 124 | os.makedirs(os.path.join(config_dir, "storage", "resources"), exist_ok=True) |
| 125 | write_config(config_dir, args.port) |
| 126 | print(f"[python] config dir: {config_dir}", flush=True) |
| 127 | |
| 128 | try: |
| 129 | reticulum = RNS.Reticulum(config_dir) |
| 130 | except Exception as e: |
| 131 | print(f"[python] Reticulum init failed: {e}", file=sys.stderr) |
| 132 | sys.exit(3) |
| 133 | |
| 134 | # Identity (fixed if --identity-file was provided and exists) |
| 135 | if args.identity_file and os.path.isfile(args.identity_file): |
| 136 | identity = RNS.Identity.from_file(args.identity_file) |
| 137 | if identity is None: |
| 138 | print("[python] failed to load identity", file=sys.stderr) |
| 139 | sys.exit(3) |
| 140 | else: |
| 141 | identity = RNS.Identity() |
| 142 | if args.identity_file: |
| 143 | identity.to_file(args.identity_file) |
| 144 | |
| 145 | destination = RNS.Destination( |
| 146 | identity, |
| 147 | RNS.Destination.IN, |
| 148 | RNS.Destination.SINGLE, |
| 149 | APP_NAME, |
| 150 | ASPECT, |
| 151 | ) |
| 152 | destination.set_link_established_callback(on_link_established) |
| 153 | destination.set_proof_strategy(RNS.Destination.PROVE_ALL) |
| 154 | |
| 155 | print(f"[python] destination hash: {destination.hash.hex()}", flush=True) |
| 156 | destination.announce() |
| 157 | print("[python] announced", flush=True) |
| 158 | |
| 159 | start = time.time() |
| 160 | last_announce = time.time() |
| 161 | while time.time() - start < args.timeout: |
| 162 | # Re-announce periodically so a sender that starts after us still |
| 163 | # has a chance to learn our destination. |
| 164 | if time.time() - last_announce >= 5.0 and not state["link"]: |
| 165 | destination.announce() |
no test coverage detected