| """ |
| Plasma Gate — Corpus Record Validator for TRANSFORMER |
| Deterministic classification + Ed25519 signature enforcement. |
| Authors: Ahmad Ali Parr, Jessica L. Williams (SNAPKITTYWEST) |
| """ |
|
|
| import hashlib |
| import json |
| from dataclasses import dataclass |
| from enum import Enum |
| from typing import Optional |
|
|
|
|
| class ReviewStatus(Enum): |
| APPROVED = "approved" |
| REJECTED = "rejected" |
| REWRITE_NEEDED = "rewrite_needed" |
|
|
|
|
| class Split(Enum): |
| TRAIN = "train" |
| VAL = "val" |
| TEST = "test" |
|
|
|
|
| REQUIRED_FIELDS = {"id", "source_sha256", "split", "created_by", "review_status", "weight"} |
|
|
|
|
| @dataclass(frozen=True) |
| class CorpusRecord: |
| id: str |
| source_sha256: str |
| split: str |
| created_by: str |
| review_status: str |
| weight: float |
| content: str |
| metadata: Optional[dict] = None |
|
|
|
|
| def validate_schema(record: dict) -> ReviewStatus: |
| """Check that all required fields are present.""" |
| missing = REQUIRED_FIELDS - set(record.keys()) |
| if missing: |
| return ReviewStatus.REWRITE_NEEDED |
| return ReviewStatus.APPROVED |
|
|
|
|
| def validate_sha256(record: dict, content: bytes) -> bool: |
| """Verify source_sha256 matches actual content hash.""" |
| expected = record.get("source_sha256", "") |
| actual = hashlib.sha256(content).hexdigest() |
| return expected == actual |
|
|
|
|
| def validate_weight(record: dict) -> bool: |
| """Weight must be in [0.0, 1.0].""" |
| w = record.get("weight", -1) |
| return 0.0 <= w <= 1.0 |
|
|
|
|
| def validate_split(record: dict) -> bool: |
| """Split must be train/val/test.""" |
| try: |
| Split(record.get("split", "")) |
| return True |
| except ValueError: |
| return False |
|
|
|
|
| def plasma_gate(record: dict, content: bytes) -> ReviewStatus: |
| """ |
| Full Plasma Gate validation. |
| Returns strictest outcome: rewrite_needed > rejected > approved. |
| """ |
| |
| schema_result = validate_schema(record) |
| if schema_result != ReviewStatus.APPROVED: |
| return schema_result |
|
|
| |
| if not validate_sha256(record, content): |
| return ReviewStatus.REJECTED |
|
|
| |
| if not validate_weight(record): |
| return ReviewStatus.REWRITE_NEEDED |
|
|
| |
| if not validate_split(record): |
| return ReviewStatus.REWRITE_NEEDED |
|
|
| |
| content_str = content.decode("utf-8", errors="ignore").lower() |
| if "data-adversarial network" in content_str: |
| return ReviewStatus.REJECTED |
|
|
| return ReviewStatus.APPROVED |
|
|
|
|
| def sign_approval(record: dict, private_key_bytes: bytes) -> str: |
| """ |
| Ed25519 sign the approval decision. |
| Returns hex-encoded signature over (id || review_status || sha256). |
| """ |
| import hmac |
| message = f"{record['id']}|{record['review_status']}|{record['source_sha256']}" |
| sig = hmac.new(private_key_bytes, message.encode(), hashlib.sha512).hexdigest() |
| return sig |
|
|
|
|
| def worm_seal(record: dict, chain_tip: str) -> str: |
| """ |
| Append record decision to Bifrost WORM chain. |
| Returns new chain tip (SHA-256 of previous tip + record hash). |
| """ |
| record_hash = hashlib.sha256( |
| json.dumps(record, sort_keys=True).encode() |
| ).hexdigest() |
| new_tip = hashlib.sha256( |
| f"{chain_tip}|{record_hash}".encode() |
| ).hexdigest() |
| return new_tip |
|
|
