#!/usr/bin/env python3
"""Gate D6: a large file crosses two daemons BY HASH, and never over the socket.
python3 -B test/test_content_e2e.py [--bootstrap host:port] [--mb 100]
[--seconds N]
One daemon registers a file; the other asks for it by content hash. The bytes
move peer-to-peer over the overlay channel, and the local socket carries a
32-byte hash and a path — docs/SWARMD-DESIGN.md §6, which is the whole reason
the content plane exists rather than a SEND loop.
⛔ IT MEASURES §6 RATHER THAN ASSERTING IT. The fetching client counts every
byte the daemon ever wrote to it. For a 100 MB transfer that number is a few
hundred; if content ever started travelling over the socket it would be a
hundred million, and this gate would say so.
⛔ IT ASSERTS ITS OWN BASELINE. If the two daemons never meet, the transfer
result says nothing about the content plane, so it exits INCONCLUSIVE.
"""
import hashlib, os, re, subprocess, sys, tempfile, time
HERE = os.path.dirname(os.path.realpath(__file__))
ROOT = os.path.dirname(HERE)
SWARMD = os.path.join(ROOT, "swarmd")
PROBE = os.path.join(ROOT, "tools", "swarmd-probe")
EXIT_PASS, EXIT_FAIL, EXIT_INCONCLUSIVE = 0, 1, 2
TOPIC = "c0" * 32
# Generous: the socket carries a handful of control frames plus progress ticks.
# A megabyte here would still be four orders of magnitude below the payload.
SOCKET_BUDGET = 256 * 1024
def arg(name, default=None):
return sys.argv[sys.argv.index(name) + 1] if name in sys.argv else default
def read(path):
try:
with open(path, "rb") as fh:
return fh.read().decode("utf-8", "replace")
except OSError:
return ""
def wait_for(path, needle, timeout, proc):
end = time.time() + timeout
while time.time() < end:
if proc is not None and proc.poll() is not None:
return False
if needle in read(path):
return True
time.sleep(0.3)
return False
def peers_seen(text):
return max([int(m) for m in re.findall(r"peers=(\d+)", text)] or [0])
def digest(path):
h = hashlib.sha256()
with open(path, "rb") as fh:
for block in iter(lambda: fh.read(1 << 20), b""):
h.update(block)
return h.hexdigest()
def main():
bootstrap = arg("--bootstrap")
megabytes = int(arg("--mb", "100"))
window = int(arg("--seconds", "120" if bootstrap else "300"))
for path in (SWARMD, PROBE):
if not os.path.isfile(path):
print("missing %s — run ./build.sh" % path)
return EXIT_INCONCLUSIVE
tmp = tempfile.mkdtemp(prefix="swarmd-content-")
asset = os.path.join(tmp, "asset.bin")
with open(asset, "wb") as fh:
# Incompressible, so nothing downstream can cheat by encoding it away.
for _ in range(megabytes):
fh.write(os.urandom(1 << 20))
want = digest(asset)
print("%d MB asset, sha256 %s" % (megabytes, want[:16]))
procs, logs = [], {}
def spawn(argv, tag):
logs[tag] = os.path.join(tmp, tag + ".log")
fh = open(logs[tag], "wb")
p = subprocess.Popen(argv, stdout=fh, stderr=subprocess.STDOUT)
procs.append(p)
return p
try:
daemons = {}
for tag in ("a", "b"):
store = os.path.join(tmp, "store-" + tag)
os.makedirs(store)
argv = [SWARMD, "--socket", os.path.join(tmp, tag + ".sock"),
"--storage", store]
if bootstrap:
argv += ["--bootstrap", bootstrap]
daemons[tag] = spawn(argv, tag)
for tag in ("a", "b"):
if not wait_for(logs[tag], "dht=up", 90, daemons[tag]):
print("\nINCONCLUSIVE: daemon %s never reached dht=up." % tag)
return EXIT_INCONCLUSIVE
print(" both daemons are on the DHT")
holder = spawn([PROBE, "--socket", os.path.join(tmp, "a.sock"),
"--topic-hex", TOPIC, "--serve-file", asset,
"--seconds", str(window)], "hold")
if not wait_for(logs["hold"], "SERVING ", 60, holder):
print("\nINCONCLUSIVE: the holder never registered the file.")
return EXIT_INCONCLUSIVE
content_hash = read(logs["hold"]).split("SERVING ")[1].split()[0]
print(" holder registered it as %s" % content_hash[:16])
dest = os.path.join(tmp, "got.bin")
fetcher = spawn([PROBE, "--socket", os.path.join(tmp, "b.sock"),
"--topic-hex", TOPIC, "--fetch-hex", content_hash,
"--dest", dest, "--seconds", str(window)], "fetch")
started = time.time()
end = started + window + 30
while time.time() < end and fetcher.poll() is None:
time.sleep(1)
if fetcher.poll() is None:
fetcher.terminate()
try: fetcher.wait(timeout=10)
except subprocess.TimeoutExpired: fetcher.kill()
out = read(logs["fetch"])
met = min(peers_seen(read(logs["a"])), peers_seen(read(logs["b"])))
if not met:
print("\nINCONCLUSIVE: the two daemons never saw each other, so the "
"transfer result says nothing about the content plane.")
return EXIT_INCONCLUSIVE
if "FETCHED " not in out:
print("\nFAIL: the daemons met and the fetch did not complete.")
print(out[-2000:])
return EXIT_FAIL
elapsed = time.time() - started
got = digest(dest)
size = os.path.getsize(dest)
socket_bytes = int(re.search(r"SOCKET-BYTES (\d+)", out).group(1))
print(" fetched %d bytes in %.1fs; the socket carried %d"
% (size, elapsed, socket_bytes))
ok = True
if size != megabytes << 20:
print("\nFAIL: %d bytes arrived, %d expected" % (size, megabytes << 20))
ok = False
if got != want:
print("\nFAIL: the bytes do not match the original")
ok = False
if socket_bytes > SOCKET_BUDGET:
print("\nFAIL: %d bytes crossed the LOCAL SOCKET for a %d MB file — "
"content is not supposed to travel that way (§6)"
% (socket_bytes, megabytes))
ok = False
if not ok:
return EXIT_FAIL
print("\nPASS: %d MB crossed the swarm by hash; the socket carried %d bytes."
% (megabytes, socket_bytes))
return EXIT_PASS
finally:
for p in procs:
if p.poll() is None:
p.terminate()
try: p.wait(timeout=5)
except subprocess.TimeoutExpired: p.kill()
print("(logs and files: %s)" % tmp)
if __name__ == "__main__":
sys.exit(main())