Spaces:
Paused
Paused
File size: 5,090 Bytes
efeddde 62711a0 efeddde 4a97a7a efeddde 4a97a7a efeddde 4a97a7a efeddde | 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 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 | import hashlib
import json
import os
import time
from pathlib import Path
from typing import Dict, Any, List, Tuple, Optional, Union
DEFAULT_CHAIN_FILE = Path("chain/blockchain.json")
def compute_block_hash(index: int, timestamp: float, previous_hash: str, evidence_record: Dict[str, Any]) -> str:
"""
Deterministically hash block contents using SHA-256.
"""
canonical_record = json.dumps(evidence_record, sort_keys=True, separators=(",", ":"))
payload = f"{index}|{timestamp:.6f}|{previous_hash}|{canonical_record}".encode("utf-8")
return hashlib.sha256(payload).hexdigest()
class LocalBlockchain:
"""
Layer 1 Local Cryptographically Linked Hash Ledger.
Provides zero-cost, high-throughput, offline-capable forensic proof of existence.
"""
def __init__(self, chain_path: Union[Path, str] = DEFAULT_CHAIN_FILE):
self.chain_path = Path(chain_path)
self.chain: List[Dict[str, Any]] = []
self._load_or_init()
def _load_or_init(self):
self.chain_path.parent.mkdir(parents=True, exist_ok=True)
if self.chain_path.exists():
try:
data = json.loads(self.chain_path.read_text(encoding="utf-8"))
if isinstance(data, list) and len(data) > 0:
self.chain = data
valid, err = self.is_valid_chain()
if not valid:
print(f"[WARN] Local blockchain integrity warning: {err}")
return
except Exception as exc:
print(f"[WARN] Failed to load local chain from {self.chain_path}: {exc}. Reinitializing.")
# Create Genesis Block
genesis = self._create_genesis_block()
self.chain = [genesis]
self._save()
def _create_genesis_block(self) -> Dict[str, Any]:
timestamp = 1726000000.0
prev_hash = "0" * 64
record = {
"title": "FaceID Genesis Block",
"protocol": "FaceID Biometric Provenance Ledger v2.0",
"status": "INITIALIZED",
}
b_hash = compute_block_hash(0, timestamp, prev_hash, record)
return {
"index": 0,
"timestamp": timestamp,
"evidence_record": record,
"previous_hash": prev_hash,
"hash": b_hash,
}
def _save(self):
self.chain_path.parent.mkdir(parents=True, exist_ok=True)
self.chain_path.write_text(
json.dumps(self.chain, indent=2, ensure_ascii=False),
encoding="utf-8"
)
def add_block(self, evidence_record: Dict[str, Any]) -> Dict[str, Any]:
"""
Append a new verified forensic record to the local blockchain ledger.
"""
import copy
latest = self.chain[-1]
index = latest["index"] + 1
timestamp = time.time()
prev_hash = latest["hash"]
sealed_record = copy.deepcopy(evidence_record)
b_hash = compute_block_hash(index, timestamp, prev_hash, sealed_record)
block = {
"index": index,
"timestamp": timestamp,
"evidence_record": sealed_record,
"previous_hash": prev_hash,
"hash": b_hash,
}
self.chain.append(block)
self._save()
return block
def is_valid_chain(self) -> Tuple[bool, Optional[str]]:
"""
Verify the mathematical integrity and continuous link hashes of the entire chain.
"""
if not self.chain:
return False, "Chain is empty"
for i in range(len(self.chain)):
block = self.chain[i]
expected_hash = compute_block_hash(
block["index"],
block["timestamp"],
block["previous_hash"],
block["evidence_record"]
)
if block["hash"] != expected_hash:
return False, f"Hash mismatch at block index {i}: expected {expected_hash}, got {block['hash']}"
if i > 0:
prev_block = self.chain[i - 1]
if block["previous_hash"] != prev_block["hash"]:
return False, f"Broken link at block index {i}: previous_hash does not match prior block hash"
return True, None
def verify_evidence_hash(self, evidence_hash: str) -> Optional[Dict[str, Any]]:
"""
Search for an evidence SHA-256 fingerprint within the verified chain.
"""
for block in self.chain:
record = block.get("evidence_record", {})
# Look for matching hash in various evidence attributes
if (
record.get("evidence_hash") == evidence_hash
or record.get("metadata_sha256") == evidence_hash
or record.get("image_sha256") == evidence_hash
or block["hash"] == evidence_hash
):
return block
return None
def get_chain(self) -> List[Dict[str, Any]]:
return self.chain
def __len__(self) -> int:
return len(self.chain)
|