File size: 3,352 Bytes
de15a9f | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 | """
CD preservation demo: scan -> self-healing .pt archive -> corrupt -> detect ->
heal -> restore bit-exact. Uses a synthetic disc image (no drive needed here).
(a) scan an image (bad-sector-tolerant)
(b) build .pt archive + restore bit-exact
(c) inject silent bit-rot (<= m shards/chunk) -> verify DETECTS it
(d) heal() repairs from survivors -> healthy again -> restore bit-exact
(e) honesty: corrupt > m shards in a chunk -> flagged LOST, never silently wrong
"""
import os
import copy
import random
import tempfile
from cdpreserve import scan, archive
random.seed(0)
print("=" * 62)
print("neural-cd-preserve -- self-healing .pt disc archive")
print("=" * 62)
# a synthetic 'disc': structured header + repeated table + random payload
disc = (b"CD-IMAGE-V1" + bytes(2037) # a 'sector'
+ (b"TABLE" * 400) * 3 # repeated -> dedup
+ os.urandom(400 * 1024))
# (a) scan (via a temp file standing in for the drive)
tmp = tempfile.NamedTemporaryFile(delete=False, suffix=".iso")
tmp.write(disc); tmp.close()
data, bad = scan.scan_image(tmp.name)
print(f"(a) scan: {len(data)//1024} KB, bad sectors: {len(bad)} "
f"-> {'PASS' if data == disc else 'FAIL'}")
# (b) build archive + restore
arc = archive.build(data, label="DEMO_DISC", k=4, m=2)
red = 6 / 4
print(f"(b) archive: {len(arc['chunks'])} unique chunks, RS 4+2 (~{red:.2f}x); "
f"restore bit-exact: {'PASS' if archive.restore(arc) == data else 'FAIL'}")
# (c) inject silent bit-rot: flip bytes in 1..2 shards per chunk
def corrupt(rec, nshards):
idxs = random.sample(range(len(rec["shards"])), nshards)
for i in idxs:
s = bytearray(rec["shards"][i]); s[0] ^= 0xFF; s[-1] ^= 0xFF
rec["shards"][i] = bytes(s) # hash now mismatches
damaged = 0
for rec in arc["chunks"].values():
n = random.randint(1, 2) # <= m, recoverable
corrupt(rec, n); damaged += n
h1 = archive.verify(arc)
print(f"(c) injected bit-rot in {damaged} shards; verify DETECTS: "
f"corrupted={h1['corrupted_shards']} repairable={h1['repairable_chunks']} "
f"lost={h1['lost_chunks']} -> {'PASS' if h1['corrupted_shards'] == damaged and h1['lost_chunks'] == 0 else 'FAIL'}")
# (d) heal + restore
repaired = archive.heal(arc)
h2 = archive.verify(arc)
restore_ok = archive.restore(arc) == data
print(f"(d) heal repaired {repaired} shards; now {'HEALTHY' if h2['healthy'] else 'DAMAGED'}; "
f"restore bit-exact: {'PASS' if restore_ok and h2['healthy'] else 'FAIL'}")
# (e) honesty: beyond RS capacity -> LOST, not silently wrong
arc2 = copy.deepcopy(arc)
victim = next(iter(arc2["chunks"].values()))
corrupt(victim, 3) # > m=2 -> unrecoverable
h3 = archive.verify(arc2)
try:
archive.restore(arc2); raised = False
except IOError:
raised = True
print(f"(e) corrupt 3 shards (> m) in one chunk -> lost={h3['lost_chunks']}, "
f"restore refuses: {'PASS (honest failure)' if h3['lost_chunks'] == 1 and raised else 'FAIL'}")
os.unlink(tmp.name)
allpass = (archive.restore(arc) == data and h1['lost_chunks'] == 0 and h2['healthy']
and restore_ok and h3['lost_chunks'] == 1 and raised)
print("=" * 62)
print(f"OVERALL: {'ALL PASS' if allpass else 'some checks failed'}")
|