mirror of
https://github.com/curly60e/pyblock.git
synced 2026-08-13 12:33:15 +02:00
Merge pull request #738 from MarcanoFilms/feature/oraclevision-integration
This commit is contained in:
Satoshi Nakamoto
GitHub
commit
27a3ddef3a
13 changed files with 1614 additions and 1 deletions
78
PR_ORACLEVISION.md
Normal file
78
PR_ORACLEVISION.md
Normal file
|
|
@ -0,0 +1,78 @@
|
|||
# Add OracleVision integration: BIP-110 spam detection and block template analysis
|
||||
|
||||
## Motivation
|
||||
|
||||
Sovereign Bitcoin node operators — especially those running **Bitcoin Knots** with **BIP-110** (`reduced_data`) policy — need local visibility into L1 spam and consensus-rule violations. Third-party block explorers and dashboards require trust. PyBLOCK already talks to `bitcoin-cli`; this PR adds **Don't Trust, Verify** tooling so operators can audit blocks and mempool composition from their own node.
|
||||
|
||||
## What was added
|
||||
|
||||
### New module: `pybitblock/oraclevision/`
|
||||
|
||||
A self-contained, zero-extra-dependency analysis engine ported from [OracleVision](https://github.com/MarcanoFilms/oraculovision):
|
||||
|
||||
| File | Purpose |
|
||||
|------|---------|
|
||||
| `script_parser.py` | BIP-110 size limits, witness/script parsing, inscription & token heuristics |
|
||||
| `bip110.py` | Per-transaction and per-block BIP-110 rule checks |
|
||||
| `spam_score.py` | 0–100 spam score and CLEAN/SUSPICIOUS/VIOLATION classification |
|
||||
| `mempool_compose.py` | `getblocktemplate` transaction categorization (economic / consolidation / coinjoin / spam) |
|
||||
| `bitcoin_cli.py` | Thin `bitcoin-cli` wrapper using PyBLOCK's existing config |
|
||||
| `config.py` | Settings loader (`config/oraclevision.conf`) |
|
||||
| `ui.py` | Terminal menus matching PyBLOCK's Rich/cyberpunk aesthetic |
|
||||
|
||||
### Menu integration
|
||||
|
||||
- **Bitcoin → OV. OracleVision** in the MONITORING section
|
||||
- Submenu:
|
||||
- **A.** BIP-110 Block Scanner (recent blocks table)
|
||||
- **B.** Mempool Glass (`getblocktemplate` categorization)
|
||||
- **C.** Block Detail View (height or hash)
|
||||
- **D.** Launch Full OracleVision TUI (if installed)
|
||||
|
||||
### Configuration
|
||||
|
||||
- `pybitblock/config/oraclevision.conf.example` — scan count, spam threshold, datadir, TUI command
|
||||
- Environment overrides for Docker/Umbrel deployments
|
||||
|
||||
### Documentation
|
||||
|
||||
- README section explaining built-in vs. full OracleVision, configuration, and how to extend detection logic
|
||||
|
||||
## Relationship to OracleVision
|
||||
|
||||
This PR does **not** port the full Textual dashboard into PyBLOCK. Instead:
|
||||
|
||||
1. **Built-in tools** give immediate value inside PyBLOCK's existing menu-driven workflow
|
||||
2. **Launch option** promotes the standalone [OracleVision](https://github.com/MarcanoFilms/oraculovision) project for operators who want DATUM mining panels, Ocean account stats, live charts, and the full rich TUI
|
||||
|
||||
The detection logic is shared in spirit with OracleVision and designed to be maintained in one place (`pybitblock/oraclevision/`) so the community can improve heuristics via PRs without touching UI code.
|
||||
|
||||
## Design principles
|
||||
|
||||
- **Low dependencies** — uses only `bitcoin-cli` (same as PyBLOCK) and existing Rich UI
|
||||
- **Modular** — detection rules separated from terminal presentation
|
||||
- **Community-extensible** — documented module boundaries for new BIP-110 checks and spam heuristics
|
||||
- **Knots + BIP-110 aligned** — version bit 4 signaling, reduced_data rule checks, local verification framing
|
||||
|
||||
## Testing notes
|
||||
|
||||
1. Requires a synced Knots/Core node with RPC enabled
|
||||
2. `getblocktemplate` needs mining RPC capability (standard on most node setups)
|
||||
3. Block scanner needs `getblock` verbosity 2 (decoded transactions)
|
||||
4. Full TUI launch requires separate OracleVision installation
|
||||
|
||||
```bash
|
||||
# Quick import check
|
||||
cd pybitblock && python3 -c "from oraclevision.bip110 import analyze_block; print('ok')"
|
||||
|
||||
# Manual test path
|
||||
python3 PyBlock.py
|
||||
# → B. Bitcoin → OV. OracleVision → A/B/C
|
||||
```
|
||||
|
||||
## Files changed
|
||||
|
||||
- `pybitblock/oraclevision/` (new package, 7 files)
|
||||
- `pybitblock/config/oraclevision.conf.example` (new)
|
||||
- `pybitblock/PyBlock.py` (menu entry + handler)
|
||||
- `README.md` (OracleVision section)
|
||||
58
README.md
58
README.md
|
|
@ -270,7 +270,63 @@
|
|||
## How to execute
|
||||
|
||||
- python3 PyBlock.py
|
||||
|
||||
|
||||
## OracleVision Integration (BIP-110 & Mempool Analysis)
|
||||
|
||||
PyBLOCK includes a lightweight integration with [OracleVision](https://github.com/MarcanoFilms/oraculovision) for sovereign node operators running **Bitcoin Knots** with **BIP-110** policy enabled. Philosophy: **Don't Trust, Verify** — all analysis runs locally against your node via `bitcoin-cli`.
|
||||
|
||||
### Built-in features (Bitcoin → OV. OracleVision)
|
||||
|
||||
| Option | What it does |
|
||||
|--------|----------------|
|
||||
| **BIP-110 Block Scanner** | Scans recent blocks for consensus violations, spam score (0–100), and status (CLEAN / SUSPICIOUS / VIOLATION) |
|
||||
| **Mempool Glass** | Categorizes your node's current `getblocktemplate` into economic, consolidation, coinjoin, and spam buckets |
|
||||
| **Block Detail View** | Deep-dive into a single block: miner tag, witness %, violation flags, problematic transactions |
|
||||
| **Launch Full OracleVision** | Opens the standalone Textual TUI if installed (DATUM mining, Ocean panels, live charts) |
|
||||
|
||||
### Configuration
|
||||
|
||||
Copy the example config and adjust for your node:
|
||||
|
||||
```bash
|
||||
cp pybitblock/config/oraclevision.conf.example pybitblock/config/oraclevision.conf
|
||||
```
|
||||
|
||||
| Setting | Default | Description |
|
||||
|---------|---------|-------------|
|
||||
| `block_scan_count` | 10 | How many recent blocks to scan |
|
||||
| `spam_score_threshold` | 45 | Score above this marks a block as VIOLATION |
|
||||
| `bitcoin_datadir` | `""` | Optional `-datadir` for bitcoin-cli |
|
||||
| `oraculovision_command` | `oraculovision` | Command to launch the full TUI |
|
||||
|
||||
Environment overrides: `ORACULOVISION_BLOCK_SCAN_COUNT`, `ORACULOVISION_SPAM_THRESHOLD`, `ORACULOVISION_COMMAND`, `BITCOIN_DATADIR`.
|
||||
|
||||
### Full OracleVision TUI (recommended for power users)
|
||||
|
||||
The built-in tools cover the essentials. For the complete dashboard — DATUM solo mining panel, Ocean account stats, live mempool charts, and navigable BIP-110 tables — install the standalone project:
|
||||
|
||||
```bash
|
||||
git clone https://github.com/MarcanoFilms/oraculovision.git
|
||||
cd oraculovision
|
||||
python -m venv .venv
|
||||
source .venv/bin/activate
|
||||
pip install -e .
|
||||
oraculovision
|
||||
```
|
||||
|
||||
From PyBLOCK, use **Bitcoin → OV. OracleVision → D. Launch Full OracleVision TUI**.
|
||||
|
||||
### Extending detection logic
|
||||
|
||||
The analysis engine lives in `pybitblock/oraclevision/` and is intentionally modular:
|
||||
|
||||
- `script_parser.py` — BIP-110 size limits and witness/script parsing
|
||||
- `bip110.py` — per-transaction and per-block rule checks
|
||||
- `spam_score.py` — heuristic scoring (community-tunable weights)
|
||||
- `mempool_compose.py` — block template categorization
|
||||
|
||||
Pull requests that improve heuristics or add new violation rules are welcome. Keep UI code in `oraclevision/ui.py` separate from detection logic.
|
||||
|
||||
|
||||
## Running PyBLOCK using Docker
|
||||
|
||||
|
|
|
|||
|
|
@ -2060,6 +2060,8 @@ def bitcoincoremenuLocal(mode): #Unified Bitcoin Core menu for local/onchain_onl
|
|||
col2.append("Mempool Monitor\n", style="white")
|
||||
col2.append(" K. ", style="bold cyan")
|
||||
col2.append("Peers Monitor\n", style="white")
|
||||
col2.append(" OV. ", style="bold cyan")
|
||||
col2.append("OracleVision\n", style="white")
|
||||
|
||||
# Tools section
|
||||
col3 = RText()
|
||||
|
|
@ -6625,6 +6627,14 @@ def bitcoincoremenuLocalControl(bcore, mode=None): #Unified Bitcoin Core local c
|
|||
print(output)
|
||||
subprocess.run(["python3", "PyVanityGenerator.py"], cwd="SPV")
|
||||
input("\a\nContinue...")
|
||||
elif bcore in ["OV", "ov"]:
|
||||
try:
|
||||
pathexec()
|
||||
from oraclevision.ui import run_oraclevision_menu
|
||||
run_oraclevision_menu(path)
|
||||
except Exception as e:
|
||||
show_error(str(e))
|
||||
logger.debug("Suppressed error: %s", e)
|
||||
else:
|
||||
if bcore.strip():
|
||||
from shared.ui import YELLOW, RESET
|
||||
|
|
|
|||
7
pybitblock/config/oraclevision.conf.example
Normal file
7
pybitblock/config/oraclevision.conf.example
Normal file
|
|
@ -0,0 +1,7 @@
|
|||
{
|
||||
"block_scan_count": 10,
|
||||
"spam_score_threshold": 45,
|
||||
"bitcoin_datadir": "",
|
||||
"oraculovision_command": "oraculovision",
|
||||
"cli_timeout_seconds": 60
|
||||
}
|
||||
22
pybitblock/oraclevision/__init__.py
Normal file
22
pybitblock/oraclevision/__init__.py
Normal file
|
|
@ -0,0 +1,22 @@
|
|||
"""
|
||||
OracleVision analysis integration for PyBLOCK.
|
||||
|
||||
Lightweight BIP-110 and mempool composition tooling for sovereign node
|
||||
operators. Core detection logic is ported from OracleVision and kept
|
||||
modular so the community can extend heuristics without touching the UI.
|
||||
|
||||
Upstream: https://github.com/MarcanoFilms/oraculovision
|
||||
"""
|
||||
|
||||
from oraclevision.bip110 import BlockAnalysis, TxAnalysis, analyze_block, analyze_transaction
|
||||
from oraclevision.mempool_compose import MempoolComposition, analyze_block_template, categorize_transaction
|
||||
|
||||
__all__ = [
|
||||
"BlockAnalysis",
|
||||
"TxAnalysis",
|
||||
"MempoolComposition",
|
||||
"analyze_block",
|
||||
"analyze_transaction",
|
||||
"analyze_block_template",
|
||||
"categorize_transaction",
|
||||
]
|
||||
320
pybitblock/oraclevision/bip110.py
Normal file
320
pybitblock/oraclevision/bip110.py
Normal file
|
|
@ -0,0 +1,320 @@
|
|||
"""
|
||||
BIP-110 block/transaction analysis engine.
|
||||
|
||||
Checks reduced_data policy rules locally against decoded block data.
|
||||
Extend _check_witness_rules() and analyze_transaction() for new rules.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass, field
|
||||
from typing import Any
|
||||
|
||||
from oraclevision.script_parser import (
|
||||
MAX_CONTROL_BLOCK_SIZE,
|
||||
MAX_OPRETURN_SIZE,
|
||||
MAX_PUSHDATA_SIZE,
|
||||
MAX_SCRIPTPUBKEY_SIZE,
|
||||
decode_coinbase_tag,
|
||||
detect_inscription_in_witness,
|
||||
detect_token_patterns,
|
||||
has_annex,
|
||||
infer_taproot_script_path,
|
||||
is_op_return,
|
||||
is_signaling_bip110,
|
||||
is_valid_taproot_control_block,
|
||||
scan_tapscript_violations,
|
||||
script_has_large_push,
|
||||
vout_script_size,
|
||||
witness_total_bytes,
|
||||
)
|
||||
from oraclevision.spam_score import classify_status, compute_spam_score
|
||||
|
||||
|
||||
@dataclass
|
||||
class TxAnalysis:
|
||||
txid: str
|
||||
weight: int
|
||||
vsize: int
|
||||
bip110_flags: set[str] = field(default_factory=set)
|
||||
signals: set[str] = field(default_factory=set)
|
||||
witness_bytes: int = 0
|
||||
|
||||
@property
|
||||
def has_bip110_violation(self) -> bool:
|
||||
return bool(self.bip110_flags)
|
||||
|
||||
@property
|
||||
def is_spam_signal(self) -> bool:
|
||||
return bool(self.signals - {"op_return"})
|
||||
|
||||
|
||||
@dataclass
|
||||
class BlockAnalysis:
|
||||
height: int
|
||||
hash: str
|
||||
miner_tag: str
|
||||
version: int
|
||||
weight: int
|
||||
tx_count: int
|
||||
bip110_signaling: bool
|
||||
spam_score: int = 0
|
||||
status: str = "CLEAN"
|
||||
violation_count: int = 0
|
||||
violation_weight: int = 0
|
||||
inscription_count: int = 0
|
||||
brc20_count: int = 0
|
||||
runes_count: int = 0
|
||||
op_return_count: int = 0
|
||||
large_witness_bytes: int = 0
|
||||
witness_pct: float = 0.0
|
||||
transactions: list[TxAnalysis] = field(default_factory=list)
|
||||
|
||||
|
||||
def _prevout_type(vin: dict) -> str | None:
|
||||
if isinstance(vin.get("prevout"), dict):
|
||||
spk = vin["prevout"].get("scriptPubKey", {})
|
||||
return spk.get("type")
|
||||
return None
|
||||
|
||||
|
||||
def _check_witness_rules(vin: dict) -> set[str]:
|
||||
flags: set[str] = set()
|
||||
witness: list[str] = vin.get("txinwitness") or vin.get("witness") or []
|
||||
if not witness:
|
||||
return flags
|
||||
|
||||
annex = has_annex(witness)
|
||||
prevout_type = _prevout_type(vin)
|
||||
is_taproot_prevout = prevout_type == "witness_v1_taproot" or prevout_type == "v1_p2tr"
|
||||
is_script_path = (
|
||||
(is_taproot_prevout and len(witness) > (2 if annex else 1))
|
||||
or infer_taproot_script_path(witness)
|
||||
)
|
||||
|
||||
if annex and (is_taproot_prevout or is_script_path):
|
||||
flags.add("taproot_annex")
|
||||
|
||||
exempt: set[int] = set()
|
||||
executing_scripts: list[int] = []
|
||||
|
||||
if annex:
|
||||
exempt.add(len(witness) - 1)
|
||||
|
||||
if is_script_path:
|
||||
cb_idx = len(witness) - (2 if annex else 1)
|
||||
tap_idx = cb_idx - 1
|
||||
exempt.add(cb_idx)
|
||||
if tap_idx >= 0:
|
||||
exempt.add(tap_idx)
|
||||
executing_scripts.append(tap_idx)
|
||||
elif is_taproot_prevout:
|
||||
sig_idx = len(witness) - 1 - (1 if annex else 0)
|
||||
if sig_idx >= 0:
|
||||
exempt.add(sig_idx)
|
||||
elif prevout_type in ("witness_v0_scripthash", "v0_p2wsh", "scripthash", "p2sh") or prevout_type is None:
|
||||
ws_idx = len(witness) - 1 - (1 if annex else 0)
|
||||
if ws_idx >= 0:
|
||||
exempt.add(ws_idx)
|
||||
executing_scripts.append(ws_idx)
|
||||
|
||||
for i, item in enumerate(witness):
|
||||
if i in exempt:
|
||||
continue
|
||||
if len(item) // 2 > MAX_PUSHDATA_SIZE:
|
||||
flags.add("large_pushdata")
|
||||
break
|
||||
|
||||
if is_script_path:
|
||||
cb_idx = len(witness) - (2 if annex else 1)
|
||||
cb = witness[cb_idx]
|
||||
if len(cb) // 2 > MAX_CONTROL_BLOCK_SIZE:
|
||||
flags.add("large_control_block")
|
||||
if is_valid_taproot_control_block(cb):
|
||||
leaf_version = int(cb[:2], 16) & 0xFE
|
||||
if leaf_version != 0xC0:
|
||||
flags.add("undefined_witness")
|
||||
tap_idx = cb_idx - 1
|
||||
if tap_idx >= 0:
|
||||
tapscript = witness[tap_idx]
|
||||
op_success, op_if = scan_tapscript_violations(tapscript)
|
||||
if op_success:
|
||||
flags.add("op_success")
|
||||
if op_if:
|
||||
flags.add("op_if_notif")
|
||||
|
||||
if "large_pushdata" not in flags:
|
||||
for idx in executing_scripts:
|
||||
if script_has_large_push(witness[idx]):
|
||||
flags.add("large_pushdata")
|
||||
break
|
||||
|
||||
return flags
|
||||
|
||||
|
||||
def _check_scriptsig_rules(vin: dict) -> set[str]:
|
||||
flags: set[str] = set()
|
||||
scriptsig = vin.get("scriptSig", {})
|
||||
hex_sig = scriptsig.get("hex", "") if isinstance(scriptsig, dict) else ""
|
||||
if not hex_sig:
|
||||
return flags
|
||||
|
||||
asm = scriptsig.get("asm", "") if isinstance(scriptsig, dict) else ""
|
||||
if asm:
|
||||
parts = asm.split()
|
||||
prevout_type = _prevout_type(vin)
|
||||
redeem_idx = len(parts) - 1 if prevout_type in ("scripthash", "p2sh") else -1
|
||||
for i, part in enumerate(parts):
|
||||
if part.startswith("OP_"):
|
||||
continue
|
||||
if i == redeem_idx:
|
||||
if script_has_large_push(part):
|
||||
flags.add("large_pushdata")
|
||||
continue
|
||||
if len(part) // 2 > MAX_PUSHDATA_SIZE:
|
||||
flags.add("large_pushdata")
|
||||
break
|
||||
elif script_has_large_push(hex_sig):
|
||||
flags.add("large_pushdata")
|
||||
return flags
|
||||
|
||||
|
||||
def analyze_transaction(tx: dict[str, Any]) -> TxAnalysis:
|
||||
txid = tx.get("txid", tx.get("hash", ""))
|
||||
weight = int(tx.get("weight") or (tx.get("vsize", 0) * 4))
|
||||
vsize = int(tx.get("vsize") or weight // 4)
|
||||
|
||||
analysis = TxAnalysis(txid=txid, weight=weight, vsize=vsize)
|
||||
bip110: set[str] = set()
|
||||
signals: set[str] = set()
|
||||
|
||||
for vout in tx.get("vout", []):
|
||||
size = vout_script_size(vout)
|
||||
if is_op_return(vout):
|
||||
signals.add("op_return")
|
||||
if size > MAX_OPRETURN_SIZE:
|
||||
bip110.add("large_scriptpubkey")
|
||||
elif size > MAX_SCRIPTPUBKEY_SIZE:
|
||||
bip110.add("large_scriptpubkey")
|
||||
|
||||
witness_bytes = 0
|
||||
all_hex = txid
|
||||
|
||||
for vin in tx.get("vin", []):
|
||||
if vin.get("coinbase"):
|
||||
continue
|
||||
witness: list[str] = vin.get("txinwitness") or vin.get("witness") or []
|
||||
witness_bytes += witness_total_bytes(witness)
|
||||
all_hex += "".join(witness)
|
||||
|
||||
bip110 |= _check_witness_rules(vin)
|
||||
bip110 |= _check_scriptsig_rules(vin)
|
||||
|
||||
if detect_inscription_in_witness(witness):
|
||||
signals.add("inscription")
|
||||
|
||||
scriptsig = vin.get("scriptSig", {})
|
||||
if isinstance(scriptsig, dict):
|
||||
all_hex += scriptsig.get("hex", "")
|
||||
|
||||
for vout in tx.get("vout", []):
|
||||
spk = vout.get("scriptPubKey", {})
|
||||
all_hex += spk.get("hex", "")
|
||||
|
||||
token_hits = detect_token_patterns(all_hex)
|
||||
signals |= token_hits
|
||||
|
||||
analysis.bip110_flags = bip110
|
||||
analysis.signals = signals
|
||||
analysis.witness_bytes = witness_bytes
|
||||
return analysis
|
||||
|
||||
|
||||
def analyze_block(
|
||||
block: dict[str, Any],
|
||||
*,
|
||||
spam_threshold: int = 45,
|
||||
) -> BlockAnalysis:
|
||||
height = int(block.get("height", 0))
|
||||
block_hash = block.get("hash", "")
|
||||
version = int(block.get("version", 0))
|
||||
weight = int(block.get("weight") or 0)
|
||||
|
||||
txs = block.get("tx", [])
|
||||
if txs and isinstance(txs[0], str):
|
||||
return BlockAnalysis(
|
||||
height=height,
|
||||
hash=block_hash,
|
||||
miner_tag="?",
|
||||
version=version,
|
||||
weight=weight,
|
||||
tx_count=len(txs),
|
||||
bip110_signaling=is_signaling_bip110(version),
|
||||
)
|
||||
|
||||
miner_tag = "unknown"
|
||||
tx_analyses: list[TxAnalysis] = []
|
||||
total_witness = 0
|
||||
|
||||
for tx in txs:
|
||||
if not isinstance(tx, dict):
|
||||
continue
|
||||
vin0 = tx.get("vin", [{}])[0]
|
||||
if vin0.get("coinbase"):
|
||||
miner_tag = decode_coinbase_tag(vin0["coinbase"])
|
||||
continue
|
||||
ta = analyze_transaction(tx)
|
||||
tx_analyses.append(ta)
|
||||
total_witness += ta.witness_bytes
|
||||
|
||||
violation_count = sum(1 for t in tx_analyses if t.has_bip110_violation)
|
||||
violation_weight = sum(t.weight for t in tx_analyses if t.has_bip110_violation)
|
||||
inscription_count = sum(1 for t in tx_analyses if "inscription" in t.signals)
|
||||
brc20_count = sum(1 for t in tx_analyses if "brc20" in t.signals)
|
||||
runes_count = sum(1 for t in tx_analyses if "runes" in t.signals)
|
||||
op_return_count = sum(1 for t in tx_analyses if "op_return" in t.signals)
|
||||
|
||||
large_witness_bytes = sum(
|
||||
t.witness_bytes for t in tx_analyses if t.witness_bytes > MAX_PUSHDATA_SIZE
|
||||
)
|
||||
|
||||
spam_score = compute_spam_score(
|
||||
block_weight=weight or 1,
|
||||
total_txs=len(tx_analyses),
|
||||
violation_weight=violation_weight,
|
||||
inscription_count=inscription_count,
|
||||
brc20_count=brc20_count,
|
||||
runes_count=runes_count,
|
||||
op_return_count=op_return_count,
|
||||
large_witness_bytes=large_witness_bytes,
|
||||
violation_count=violation_count,
|
||||
)
|
||||
status = classify_status(
|
||||
spam_score,
|
||||
violation_count,
|
||||
violation_weight,
|
||||
weight or 1,
|
||||
spam_threshold=spam_threshold,
|
||||
)
|
||||
witness_pct = (total_witness / max(weight, 1)) * 100 if weight else 0.0
|
||||
|
||||
return BlockAnalysis(
|
||||
height=height,
|
||||
hash=block_hash,
|
||||
miner_tag=miner_tag,
|
||||
version=version,
|
||||
weight=weight,
|
||||
tx_count=len(tx_analyses),
|
||||
bip110_signaling=is_signaling_bip110(version),
|
||||
spam_score=spam_score,
|
||||
status=status,
|
||||
violation_count=violation_count,
|
||||
violation_weight=violation_weight,
|
||||
inscription_count=inscription_count,
|
||||
brc20_count=brc20_count,
|
||||
runes_count=runes_count,
|
||||
op_return_count=op_return_count,
|
||||
large_witness_bytes=large_witness_bytes,
|
||||
witness_pct=witness_pct,
|
||||
transactions=tx_analyses,
|
||||
)
|
||||
126
pybitblock/oraclevision/bitcoin_cli.py
Normal file
126
pybitblock/oraclevision/bitcoin_cli.py
Normal file
|
|
@ -0,0 +1,126 @@
|
|||
"""
|
||||
bitcoin-cli wrapper for OracleVision analysis inside PyBLOCK.
|
||||
|
||||
Uses the same bitcoin-cli path configured in bclock.conf. No extra deps.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import os
|
||||
import subprocess
|
||||
from typing import Any
|
||||
|
||||
from oraclevision.security import resolve_bitcoin_cli, validate_rpc_method, validate_safe_path_token
|
||||
|
||||
|
||||
class BitcoinCLIError(Exception):
|
||||
"""Raised when bitcoin-cli fails or is unavailable."""
|
||||
|
||||
def __init__(self, message: str, *, hint: str | None = None) -> None:
|
||||
self.hint = hint
|
||||
super().__init__(message)
|
||||
|
||||
|
||||
class BitcoinCLI:
|
||||
"""Thin wrapper around bitcoin-cli JSON-RPC for analysis commands."""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
cli_path: str,
|
||||
datadir: str | None = None,
|
||||
timeout: float = 60.0,
|
||||
) -> None:
|
||||
try:
|
||||
self.cli_path = resolve_bitcoin_cli(cli_path or os.environ.get("BITCOIN_CLI", "bitcoin-cli"))
|
||||
self.datadir = validate_safe_path_token(
|
||||
datadir or os.environ.get("BITCOIN_DATADIR") or "",
|
||||
name="datadir",
|
||||
allow_empty=True,
|
||||
)
|
||||
except ValueError as exc:
|
||||
raise BitcoinCLIError(str(exc)) from exc
|
||||
self.timeout = timeout
|
||||
|
||||
def _base_cmd(self) -> list[str]:
|
||||
cmd = [self.cli_path]
|
||||
if self.datadir:
|
||||
cmd.extend(["-datadir", self.datadir])
|
||||
return cmd
|
||||
|
||||
def call(self, method: str, *params: Any) -> Any:
|
||||
try:
|
||||
method = validate_rpc_method(method)
|
||||
except ValueError as exc:
|
||||
raise BitcoinCLIError(str(exc)) from exc
|
||||
|
||||
cmd = self._base_cmd() + [method]
|
||||
for param in params:
|
||||
if isinstance(param, (dict, list)):
|
||||
cmd.append(json.dumps(param))
|
||||
elif isinstance(param, bool):
|
||||
cmd.append("true" if param else "false")
|
||||
else:
|
||||
cmd.append(str(param))
|
||||
|
||||
try:
|
||||
# nosemgrep: python.lang.security.audit.dangerous-subprocess-use-audit
|
||||
result = subprocess.run(
|
||||
cmd,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=self.timeout,
|
||||
check=False,
|
||||
)
|
||||
except subprocess.TimeoutExpired as exc:
|
||||
raise BitcoinCLIError(
|
||||
f"Timeout calling {method} ({self.timeout}s)",
|
||||
hint="The node may be busy or unresponsive.",
|
||||
) from exc
|
||||
except FileNotFoundError as exc:
|
||||
raise BitcoinCLIError(
|
||||
f"bitcoin-cli not found: {self.cli_path}",
|
||||
hint="Check your Knots/Core installation.",
|
||||
) from exc
|
||||
|
||||
if result.returncode != 0:
|
||||
stderr = (result.stderr or result.stdout or "").strip()
|
||||
hint = None
|
||||
lower = stderr.lower()
|
||||
if "could not connect" in lower or "connection refused" in lower:
|
||||
hint = "Start bitcoind/knots and check RPC (bitcoin.conf)."
|
||||
elif "verifying blocks" in lower or "initial block download" in lower:
|
||||
hint = "Node still syncing. Wait for IBD to finish."
|
||||
elif "not available" in lower and method == "getblocktemplate":
|
||||
hint = "Enable mining RPC or use a node that supports getblocktemplate."
|
||||
raise BitcoinCLIError(stderr or f"Error in {method}", hint=hint)
|
||||
|
||||
stdout = result.stdout.strip()
|
||||
if not stdout:
|
||||
return None
|
||||
try:
|
||||
return json.loads(stdout)
|
||||
except json.JSONDecodeError:
|
||||
return stdout
|
||||
|
||||
def get_block_count(self) -> int:
|
||||
return int(self.call("getblockcount"))
|
||||
|
||||
def get_block_hash(self, height: int) -> str:
|
||||
return str(self.call("getblockhash", height))
|
||||
|
||||
def get_block(self, block_hash: str, verbosity: int = 2) -> dict[str, Any]:
|
||||
return self.call("getblock", block_hash, verbosity)
|
||||
|
||||
def get_mempool_info(self) -> dict[str, Any]:
|
||||
return self.call("getmempoolinfo")
|
||||
|
||||
def get_block_template(self) -> dict[str, Any]:
|
||||
return self.call("getblocktemplate", {"rules": ["segwit"]})
|
||||
|
||||
def decode_raw_transaction(self, hex_data: str) -> dict[str, Any]:
|
||||
return self.call("decoderawtransaction", hex_data)
|
||||
|
||||
@classmethod
|
||||
def from_path_config(cls, path: dict[str, str], datadir: str = "", timeout: float = 60.0) -> "BitcoinCLI":
|
||||
return cls(path.get("bitcoincli", "bitcoin-cli"), datadir=datadir, timeout=timeout)
|
||||
107
pybitblock/oraclevision/config.py
Normal file
107
pybitblock/oraclevision/config.py
Normal file
|
|
@ -0,0 +1,107 @@
|
|||
"""
|
||||
OracleVision settings for PyBLOCK.
|
||||
|
||||
Stored in config/oraclevision.conf (JSON). Environment overrides:
|
||||
ORACULOVISION_BLOCK_SCAN_COUNT
|
||||
ORACULOVISION_SPAM_THRESHOLD
|
||||
ORACULOVISION_COMMAND
|
||||
BITCOIN_DATADIR
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import os
|
||||
from dataclasses import dataclass
|
||||
|
||||
from config import cfg
|
||||
from oraclevision.security import validate_safe_path_token
|
||||
|
||||
|
||||
_DEFAULTS = {
|
||||
"block_scan_count": 10,
|
||||
"spam_score_threshold": 45,
|
||||
"bitcoin_datadir": "",
|
||||
"oraculovision_command": "oraculovision",
|
||||
"cli_timeout_seconds": 60,
|
||||
}
|
||||
|
||||
|
||||
@dataclass
|
||||
class OracleVisionSettings:
|
||||
block_scan_count: int = 10
|
||||
spam_score_threshold: int = 45
|
||||
bitcoin_datadir: str = ""
|
||||
oraculovision_command: str = "oraculovision"
|
||||
cli_timeout_seconds: int = 60
|
||||
load_error: str | None = None
|
||||
|
||||
|
||||
def _safe_int(value: object, default: int, *, field: str, errors: list[str]) -> int:
|
||||
try:
|
||||
return int(value) # type: ignore[arg-type]
|
||||
except (TypeError, ValueError):
|
||||
errors.append(f"Invalid {field}; using default {default}")
|
||||
return default
|
||||
|
||||
|
||||
def load_settings() -> OracleVisionSettings:
|
||||
"""Load OracleVision config, merging defaults, file, and env vars."""
|
||||
data = dict(_DEFAULTS)
|
||||
errors: list[str] = []
|
||||
filepath = os.path.join(cfg.config_dir, "oraclevision.conf")
|
||||
|
||||
if os.path.isfile(filepath):
|
||||
try:
|
||||
with open(filepath, "r", encoding="utf-8") as f:
|
||||
file_data = json.load(f)
|
||||
if isinstance(file_data, dict):
|
||||
data.update(file_data)
|
||||
else:
|
||||
errors.append("oraclevision.conf must be a JSON object; using defaults")
|
||||
except json.JSONDecodeError as exc:
|
||||
errors.append(f"Invalid JSON in oraclevision.conf: {exc}")
|
||||
except OSError as exc:
|
||||
errors.append(f"Could not read oraclevision.conf: {exc}")
|
||||
|
||||
if env_count := os.environ.get("ORACULOVISION_BLOCK_SCAN_COUNT"):
|
||||
data["block_scan_count"] = _safe_int(env_count, data["block_scan_count"], field="block_scan_count", errors=errors)
|
||||
if env_threshold := os.environ.get("ORACULOVISION_SPAM_THRESHOLD"):
|
||||
data["spam_score_threshold"] = _safe_int(
|
||||
env_threshold, data["spam_score_threshold"], field="spam_score_threshold", errors=errors
|
||||
)
|
||||
if env_cmd := os.environ.get("ORACULOVISION_COMMAND"):
|
||||
data["oraculovision_command"] = env_cmd
|
||||
if env_datadir := os.environ.get("BITCOIN_DATADIR"):
|
||||
data["bitcoin_datadir"] = env_datadir
|
||||
|
||||
bitcoin_datadir = ""
|
||||
try:
|
||||
bitcoin_datadir = validate_safe_path_token(
|
||||
str(data.get("bitcoin_datadir", "")), name="bitcoin_datadir", allow_empty=True
|
||||
)
|
||||
except ValueError as exc:
|
||||
errors.append(str(exc))
|
||||
bitcoin_datadir = ""
|
||||
|
||||
oraculovision_command = str(data.get("oraculovision_command", "oraculovision"))
|
||||
try:
|
||||
validate_safe_path_token(oraculovision_command, name="oraculovision_command", allow_empty=False)
|
||||
except ValueError as exc:
|
||||
errors.append(str(exc))
|
||||
oraculovision_command = _DEFAULTS["oraculovision_command"]
|
||||
|
||||
return OracleVisionSettings(
|
||||
block_scan_count=_safe_int(
|
||||
data.get("block_scan_count", 10), 10, field="block_scan_count", errors=errors
|
||||
),
|
||||
spam_score_threshold=_safe_int(
|
||||
data.get("spam_score_threshold", 45), 45, field="spam_score_threshold", errors=errors
|
||||
),
|
||||
bitcoin_datadir=bitcoin_datadir,
|
||||
oraculovision_command=oraculovision_command,
|
||||
cli_timeout_seconds=_safe_int(
|
||||
data.get("cli_timeout_seconds", 60), 60, field="cli_timeout_seconds", errors=errors
|
||||
),
|
||||
load_error="; ".join(errors) if errors else None,
|
||||
)
|
||||
166
pybitblock/oraclevision/mempool_compose.py
Normal file
166
pybitblock/oraclevision/mempool_compose.py
Normal file
|
|
@ -0,0 +1,166 @@
|
|||
"""
|
||||
Block template composition analysis for Mempool Glass.
|
||||
|
||||
Classifies transactions from getblocktemplate into economic, consolidation,
|
||||
coinjoin, and spam buckets. Extend categorize_transaction() to add categories.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass
|
||||
from typing import Any, Callable
|
||||
|
||||
from oraclevision.bip110 import analyze_transaction
|
||||
from oraclevision.script_parser import MAX_PUSHDATA_SIZE, witness_total_bytes
|
||||
|
||||
|
||||
@dataclass
|
||||
class MempoolComposition:
|
||||
"""Composition stats derived from the node's block template."""
|
||||
|
||||
total_tx: int = 0
|
||||
total_weight: int = 0
|
||||
analyzed_tx: int = 0
|
||||
analyzed_weight: int = 0
|
||||
economic_weight: int = 0
|
||||
consolidation_weight: int = 0
|
||||
coinjoin_weight: int = 0
|
||||
spam_weight: int = 0
|
||||
economic_count: int = 0
|
||||
consolidation_count: int = 0
|
||||
coinjoin_count: int = 0
|
||||
spam_count: int = 0
|
||||
template_height: int = 0
|
||||
weight_limit: int = 4_000_000
|
||||
fill_pct: float = 0.0
|
||||
mempool_size: int = 0
|
||||
source: str = "block_template"
|
||||
error: str | None = None
|
||||
|
||||
def pct(self, weight: int) -> float:
|
||||
base = self.analyzed_weight or 1
|
||||
return (weight / base) * 100
|
||||
|
||||
|
||||
def _witness_has_oversized_item(tx: dict[str, Any]) -> bool:
|
||||
for vin in tx.get("vin", []):
|
||||
witness = vin.get("txinwitness") or vin.get("witness") or []
|
||||
for item in witness:
|
||||
if len(item) // 2 > MAX_PUSHDATA_SIZE:
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
def _excess_witness_ratio(tx: dict[str, Any]) -> bool:
|
||||
weight = int(tx.get("weight") or 1)
|
||||
wbytes = 0
|
||||
for vin in tx.get("vin", []):
|
||||
witness = vin.get("txinwitness") or vin.get("witness") or []
|
||||
wbytes += witness_total_bytes(witness)
|
||||
return wbytes > 2000 and (wbytes / weight) > 0.45
|
||||
|
||||
|
||||
def _is_consolidation(tx: dict[str, Any]) -> bool:
|
||||
vin = len(tx.get("vin", []))
|
||||
vout = len(tx.get("vout", []))
|
||||
return vout <= 2 and vin >= 5
|
||||
|
||||
|
||||
def _is_coinjoin(tx: dict[str, Any]) -> bool:
|
||||
vin = tx.get("vin", [])
|
||||
vout = tx.get("vout", [])
|
||||
if len(vin) < 5 or len(vout) < 5:
|
||||
return False
|
||||
in_vals: dict[float, int] = {}
|
||||
out_vals: dict[float, int] = {}
|
||||
for i in vin:
|
||||
v = (i.get("prevout") or {}).get("value")
|
||||
if v is not None:
|
||||
in_vals[v] = in_vals.get(v, 0) + 1
|
||||
for o in vout:
|
||||
v = o.get("value")
|
||||
if v is not None:
|
||||
out_vals[v] = out_vals.get(v, 0) + 1
|
||||
if not in_vals and not out_vals:
|
||||
return len(vin) >= 8 and len(vout) >= 8 and abs(len(vin) - len(vout)) <= 2
|
||||
unique = len(set(in_vals) | set(out_vals))
|
||||
total = len(vin) + len(vout)
|
||||
return unique <= total // 2
|
||||
|
||||
|
||||
def _is_spam_tx(tx: dict[str, Any]) -> bool:
|
||||
analysis = analyze_transaction(tx)
|
||||
if analysis.has_bip110_violation or analysis.is_spam_signal:
|
||||
return True
|
||||
if _witness_has_oversized_item(tx):
|
||||
return True
|
||||
if _excess_witness_ratio(tx):
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
def categorize_transaction(tx: dict[str, Any]) -> str:
|
||||
"""Return category: spam, coinjoin, consolidation, economic."""
|
||||
if _is_spam_tx(tx):
|
||||
return "spam"
|
||||
if _is_coinjoin(tx):
|
||||
return "coinjoin"
|
||||
if _is_consolidation(tx):
|
||||
return "consolidation"
|
||||
return "economic"
|
||||
|
||||
|
||||
def analyze_block_template(
|
||||
template: dict[str, Any],
|
||||
decode_tx: Callable[[str], dict[str, Any]],
|
||||
) -> MempoolComposition:
|
||||
"""Classify all transactions in the node's current block template."""
|
||||
result = MempoolComposition()
|
||||
txs = template.get("transactions", [])
|
||||
result.template_height = int(template.get("height", 0))
|
||||
result.weight_limit = int(template.get("weightlimit", 4_000_000))
|
||||
result.total_tx = len(txs)
|
||||
|
||||
if not txs:
|
||||
result.error = "Block template is empty"
|
||||
return result
|
||||
|
||||
for entry in txs:
|
||||
weight = int(entry.get("weight", 0))
|
||||
result.total_weight += weight
|
||||
hex_data = entry.get("data", "")
|
||||
if not hex_data:
|
||||
continue
|
||||
try:
|
||||
tx = decode_tx(hex_data)
|
||||
if entry.get("txid"):
|
||||
tx["txid"] = entry["txid"]
|
||||
if weight and not tx.get("weight"):
|
||||
tx["weight"] = weight
|
||||
except Exception:
|
||||
continue
|
||||
|
||||
result.analyzed_tx += 1
|
||||
w = int(tx.get("weight") or weight or 0)
|
||||
result.analyzed_weight += w
|
||||
|
||||
cat = categorize_transaction(tx)
|
||||
if cat == "spam":
|
||||
result.spam_weight += w
|
||||
result.spam_count += 1
|
||||
elif cat == "coinjoin":
|
||||
result.coinjoin_weight += w
|
||||
result.coinjoin_count += 1
|
||||
elif cat == "consolidation":
|
||||
result.consolidation_weight += w
|
||||
result.consolidation_count += 1
|
||||
else:
|
||||
result.economic_weight += w
|
||||
result.economic_count += 1
|
||||
|
||||
if result.analyzed_tx == 0:
|
||||
result.error = "Could not decode transactions from the template"
|
||||
else:
|
||||
result.fill_pct = (result.analyzed_weight / result.weight_limit * 100) if result.weight_limit else 0
|
||||
|
||||
return result
|
||||
239
pybitblock/oraclevision/script_parser.py
Normal file
239
pybitblock/oraclevision/script_parser.py
Normal file
|
|
@ -0,0 +1,239 @@
|
|||
"""
|
||||
Low-level Bitcoin script/witness parsing helpers.
|
||||
|
||||
Ported from OracleVision (https://github.com/MarcanoFilms/oraculovision).
|
||||
These functions implement BIP-110 size checks and spam heuristics. Extend
|
||||
this module when adding new detection rules — keep UI code separate.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
from typing import Iterable
|
||||
|
||||
# BIP-110 size limits (reduced_data policy)
|
||||
MAX_SCRIPTPUBKEY_SIZE = 34
|
||||
MAX_OPRETURN_SIZE = 83
|
||||
MAX_PUSHDATA_SIZE = 256
|
||||
MAX_CONTROL_BLOCK_SIZE = 257
|
||||
|
||||
OP_IF = 0x63
|
||||
OP_NOTIF = 0x64
|
||||
OP_FALSE = 0x00
|
||||
|
||||
_SPAM_HEX_PATTERNS = (
|
||||
b"6272632d3230", # brc-20
|
||||
b"2270223a22627263", # "p":"brc
|
||||
b"7469636b", # tick
|
||||
b"6f7264", # ord
|
||||
b"52554e45", # RUNE
|
||||
b"746578742f706c61696e", # text/plain
|
||||
)
|
||||
|
||||
|
||||
def hex_to_bytes(hex_str: str) -> bytes:
|
||||
if not hex_str:
|
||||
return b""
|
||||
try:
|
||||
return bytes.fromhex(hex_str)
|
||||
except ValueError:
|
||||
return b""
|
||||
|
||||
|
||||
def witness_total_bytes(witness: Iterable[str] | None) -> int:
|
||||
if not witness:
|
||||
return 0
|
||||
return sum(len(w) // 2 for w in witness)
|
||||
|
||||
|
||||
def has_annex(witness: list[str] | None) -> bool:
|
||||
if not witness or len(witness) < 2:
|
||||
return False
|
||||
return witness[-1].startswith("50")
|
||||
|
||||
|
||||
def is_valid_taproot_control_block(cb_hex: str) -> bool:
|
||||
cb = hex_to_bytes(cb_hex)
|
||||
if len(cb) < 33 or (len(cb) - 33) % 32 != 0:
|
||||
return False
|
||||
return (cb[0] & 0xFE) >= 0xC0
|
||||
|
||||
|
||||
def infer_taproot_script_path(witness: list[str]) -> bool:
|
||||
"""Infer taproot script-path spend from witness structure without prevout."""
|
||||
if not witness or len(witness) < 3:
|
||||
return False
|
||||
annex = has_annex(witness)
|
||||
cb_idx = len(witness) - (2 if annex else 1)
|
||||
if cb_idx < 1:
|
||||
return False
|
||||
return is_valid_taproot_control_block(witness[cb_idx])
|
||||
|
||||
|
||||
def script_has_large_push(script_hex: str) -> bool:
|
||||
"""BIP-110 Rule 2: OP_PUSHDATA payloads > 256 bytes inside executing scripts."""
|
||||
buf = hex_to_bytes(script_hex)
|
||||
i = 0
|
||||
while i < len(buf):
|
||||
op = buf[i]
|
||||
if 0x01 <= op <= 0x4B:
|
||||
header_len, data_len = 1, op
|
||||
elif op == 0x4C:
|
||||
if i + 2 > len(buf):
|
||||
break
|
||||
header_len, data_len = 2, buf[i + 1]
|
||||
elif op == 0x4D:
|
||||
if i + 3 > len(buf):
|
||||
break
|
||||
header_len, data_len = 3, int.from_bytes(buf[i + 1 : i + 3], "little")
|
||||
elif op == 0x4E:
|
||||
if i + 5 > len(buf):
|
||||
break
|
||||
header_len, data_len = 5, int.from_bytes(buf[i + 1 : i + 5], "little")
|
||||
else:
|
||||
i += 1
|
||||
continue
|
||||
if i + header_len + data_len > len(buf):
|
||||
break
|
||||
if data_len > MAX_PUSHDATA_SIZE:
|
||||
return True
|
||||
i += header_len + data_len
|
||||
return False
|
||||
|
||||
|
||||
def scan_tapscript_violations(script_hex: str) -> tuple[bool, bool]:
|
||||
"""Return (op_success, op_if_notif) for BIP-110 rules 6 & 7."""
|
||||
buf = hex_to_bytes(script_hex)
|
||||
op_success = False
|
||||
op_if_notif = False
|
||||
i = 0
|
||||
while i < len(buf):
|
||||
op = buf[i]
|
||||
if 0x01 <= op <= 0x4B:
|
||||
i += 1 + op
|
||||
continue
|
||||
if op == 0x4C:
|
||||
if i + 1 >= len(buf):
|
||||
break
|
||||
i += 2 + buf[i + 1]
|
||||
continue
|
||||
if op == 0x4D:
|
||||
if i + 2 >= len(buf):
|
||||
break
|
||||
i += 3 + int.from_bytes(buf[i + 1 : i + 3], "little")
|
||||
continue
|
||||
if op == 0x4E:
|
||||
if i + 4 >= len(buf):
|
||||
break
|
||||
i += 5 + int.from_bytes(buf[i + 1 : i + 5], "little")
|
||||
continue
|
||||
|
||||
if (
|
||||
op in (80, 98)
|
||||
or (126 <= op <= 129)
|
||||
or (131 <= op <= 134)
|
||||
or (137 <= op <= 138)
|
||||
or (141 <= op <= 142)
|
||||
or (149 <= op <= 153)
|
||||
or (187 <= op <= 254)
|
||||
):
|
||||
op_success = True
|
||||
elif op in (OP_IF, OP_NOTIF):
|
||||
op_if_notif = True
|
||||
|
||||
i += 1
|
||||
if op_success and op_if_notif:
|
||||
break
|
||||
return op_success, op_if_notif
|
||||
|
||||
|
||||
def has_op_false_op_if_envelope(script_hex: str) -> bool:
|
||||
"""Detect Ordinals inscription envelope: OP_FALSE ... OP_IF."""
|
||||
buf = hex_to_bytes(script_hex)
|
||||
if len(buf) < 3:
|
||||
return False
|
||||
for i in range(len(buf) - 1):
|
||||
if buf[i] == OP_FALSE and buf[i + 1] == OP_IF:
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
def detect_inscription_in_witness(witness: list[str] | None) -> bool:
|
||||
if not witness:
|
||||
return False
|
||||
annex = has_annex(witness)
|
||||
if len(witness) > (2 if annex else 1):
|
||||
script_idx = len(witness) - (3 if annex else 2)
|
||||
if script_idx >= 0:
|
||||
tapscript = witness[script_idx]
|
||||
if has_op_false_op_if_envelope(tapscript):
|
||||
return True
|
||||
for item in witness:
|
||||
if has_op_false_op_if_envelope(item):
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
def detect_token_patterns(hex_blob: str) -> set[str]:
|
||||
"""Heuristic detection of BRC-20, Runes, Ordinals content in hex."""
|
||||
found: set[str] = set()
|
||||
raw = hex_to_bytes(hex_blob)
|
||||
ascii_text = raw.decode("ascii", errors="ignore").lower()
|
||||
|
||||
if b"6272632d3230" in raw or b'"p":"brc-20"' in raw or b'"p": "brc-20"' in raw:
|
||||
found.add("brc20")
|
||||
if b"7469636b" in raw and (b"627263" in raw or b"6f7264" in raw):
|
||||
found.add("brc20")
|
||||
if b"52554e45" in raw:
|
||||
found.add("runes")
|
||||
if b"6f7264" in raw or b"746578742f706c61696e" in raw:
|
||||
found.add("ordinals")
|
||||
if "ord" in ascii_text or "inscription" in ascii_text:
|
||||
found.add("ordinals")
|
||||
|
||||
for pat in _SPAM_HEX_PATTERNS:
|
||||
if pat in raw:
|
||||
if pat == b"52554e45":
|
||||
found.add("runes")
|
||||
elif pat in (b"6272632d3230", b"2270223a22627263", b"7469636b"):
|
||||
found.add("brc20")
|
||||
else:
|
||||
found.add("ordinals")
|
||||
return found
|
||||
|
||||
|
||||
def decode_coinbase_tag(coinbase_hex: str) -> str:
|
||||
"""Extract readable miner/pool tag from coinbase hex."""
|
||||
raw = hex_to_bytes(coinbase_hex)
|
||||
if len(raw) < 4:
|
||||
return "unknown"
|
||||
|
||||
text = raw.decode("ascii", errors="ignore")
|
||||
runs = re.findall(r"[\x20-\x7e]{4,}", text)
|
||||
if not runs:
|
||||
return "unknown"
|
||||
|
||||
pool_runs = [r.strip() for r in runs if "/" in r and len(r.strip()) >= 5]
|
||||
if pool_runs:
|
||||
return max(pool_runs, key=len)[:40]
|
||||
|
||||
candidates = [r.strip() for r in runs if len(r.strip()) >= 6]
|
||||
if candidates:
|
||||
return max(candidates, key=len)[:40]
|
||||
return runs[-1].strip()[:40] or "unknown"
|
||||
|
||||
|
||||
def is_signaling_bip110(version: int) -> bool:
|
||||
"""BIP-110 reduced_data uses version bit 4."""
|
||||
return bool(version & (1 << 4))
|
||||
|
||||
|
||||
def vout_script_size(vout: dict) -> int:
|
||||
spk = vout.get("scriptPubKey", {})
|
||||
hex_data = spk.get("hex", "")
|
||||
return len(hex_data) // 2
|
||||
|
||||
|
||||
def is_op_return(vout: dict) -> bool:
|
||||
spk = vout.get("scriptPubKey", {})
|
||||
return spk.get("type") == "nulldata" or spk.get("asm", "").startswith("OP_RETURN")
|
||||
74
pybitblock/oraclevision/security.py
Normal file
74
pybitblock/oraclevision/security.py
Normal file
|
|
@ -0,0 +1,74 @@
|
|||
"""
|
||||
Input validation helpers for OracleVision subprocess and config paths.
|
||||
|
||||
Prevents command injection when launching external binaries configured by
|
||||
the node operator (bitcoin-cli path, oraculovision command, datadir).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import re
|
||||
import shlex
|
||||
import shutil
|
||||
|
||||
_SHELL_META = re.compile(r"[;|&$`<>\"'\n\\]")
|
||||
_RPC_METHOD = re.compile(r"^[a-z][a-z0-9_]*$", re.I)
|
||||
|
||||
|
||||
def validate_safe_path_token(value: str, *, name: str = "path", allow_empty: bool = True) -> str:
|
||||
"""Reject shell metacharacters in filesystem path tokens."""
|
||||
value = (value or "").strip()
|
||||
if not value:
|
||||
if allow_empty:
|
||||
return ""
|
||||
raise ValueError(f"{name} must not be empty")
|
||||
if _SHELL_META.search(value):
|
||||
raise ValueError(f"Invalid characters in {name}")
|
||||
return value
|
||||
|
||||
|
||||
def validate_rpc_method(method: str) -> str:
|
||||
"""Ensure bitcoin-cli RPC method names are safe tokens."""
|
||||
method = (method or "").strip()
|
||||
if not _RPC_METHOD.fullmatch(method):
|
||||
raise ValueError(f"Invalid RPC method: {method!r}")
|
||||
return method
|
||||
|
||||
|
||||
def resolve_executable(command: str) -> list[str]:
|
||||
"""Resolve a single executable name or absolute path for subprocess.run."""
|
||||
command = (command or "").strip()
|
||||
if not command:
|
||||
raise ValueError("Empty command")
|
||||
|
||||
parts = shlex.split(command)
|
||||
if len(parts) != 1:
|
||||
raise ValueError("Command must be a single executable (no shell arguments)")
|
||||
|
||||
exe = validate_safe_path_token(parts[0], name="command", allow_empty=False)
|
||||
|
||||
if os.path.isabs(exe):
|
||||
if not os.path.isfile(exe) or not os.access(exe, os.X_OK):
|
||||
raise ValueError(f"Not executable: {exe}")
|
||||
return [exe]
|
||||
|
||||
resolved = shutil.which(exe)
|
||||
if not resolved:
|
||||
raise ValueError(f"Command not found: {exe}")
|
||||
return [resolved]
|
||||
|
||||
|
||||
def resolve_bitcoin_cli(cli_path: str) -> str:
|
||||
"""Resolve and validate bitcoin-cli executable path."""
|
||||
cli_path = validate_safe_path_token(cli_path or "bitcoin-cli", name="bitcoincli", allow_empty=False)
|
||||
|
||||
if os.path.isabs(cli_path):
|
||||
if not os.path.isfile(cli_path):
|
||||
raise ValueError(f"bitcoin-cli not found: {cli_path}")
|
||||
return cli_path
|
||||
|
||||
resolved = shutil.which(cli_path)
|
||||
if not resolved:
|
||||
raise ValueError(f"bitcoin-cli not found: {cli_path}")
|
||||
return resolved
|
||||
74
pybitblock/oraclevision/spam_score.py
Normal file
74
pybitblock/oraclevision/spam_score.py
Normal file
|
|
@ -0,0 +1,74 @@
|
|||
"""
|
||||
Spam score and BIP-110 status classification.
|
||||
|
||||
Weights are heuristic — tune in oraclevision.conf or extend compute_spam_score()
|
||||
for community-driven improvements.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
|
||||
def compute_spam_score(
|
||||
*,
|
||||
block_weight: int,
|
||||
total_txs: int,
|
||||
violation_weight: int,
|
||||
inscription_count: int,
|
||||
brc20_count: int,
|
||||
runes_count: int,
|
||||
op_return_count: int,
|
||||
large_witness_bytes: int,
|
||||
violation_count: int,
|
||||
) -> int:
|
||||
"""Compute 0-100 spam score for a block."""
|
||||
if block_weight <= 0:
|
||||
block_weight = 1
|
||||
if total_txs <= 0:
|
||||
total_txs = 1
|
||||
|
||||
violation_ratio = violation_weight / block_weight
|
||||
inscription_ratio = inscription_count / total_txs
|
||||
token_ratio = (brc20_count + runes_count) / total_txs
|
||||
witness_ratio = large_witness_bytes / block_weight
|
||||
op_return_ratio = op_return_count / total_txs
|
||||
|
||||
score = (
|
||||
40 * violation_ratio
|
||||
+ 25 * inscription_ratio
|
||||
+ 15 * witness_ratio
|
||||
+ 10 * op_return_ratio
|
||||
+ 10 * token_ratio * 5
|
||||
)
|
||||
|
||||
if violation_count > 10:
|
||||
score += min(20, violation_count)
|
||||
|
||||
return min(100, int(round(score)))
|
||||
|
||||
|
||||
def classify_status(
|
||||
spam_score: int,
|
||||
violation_count: int,
|
||||
violation_weight: int,
|
||||
block_weight: int,
|
||||
*,
|
||||
spam_threshold: int = 45,
|
||||
violation_pct_threshold: float = 5.0,
|
||||
) -> str:
|
||||
"""Return CLEAN, SUSPICIOUS, or VIOLATION."""
|
||||
violation_pct = (violation_weight / max(block_weight, 1)) * 100
|
||||
|
||||
if spam_score > spam_threshold or violation_pct > violation_pct_threshold:
|
||||
return "VIOLATION"
|
||||
if spam_score >= 15 or violation_count > 0:
|
||||
return "SUSPICIOUS"
|
||||
return "CLEAN"
|
||||
|
||||
|
||||
def status_style(status: str) -> str:
|
||||
"""Rich style name for terminal display."""
|
||||
return {
|
||||
"CLEAN": "bold green",
|
||||
"SUSPICIOUS": "bold yellow",
|
||||
"VIOLATION": "bold red",
|
||||
}.get(status, "white")
|
||||
334
pybitblock/oraclevision/ui.py
Normal file
334
pybitblock/oraclevision/ui.py
Normal file
|
|
@ -0,0 +1,334 @@
|
|||
"""
|
||||
Terminal UI for OracleVision features inside PyBLOCK.
|
||||
|
||||
Don't Trust, Verify — all analysis runs locally against your Knots node.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import subprocess
|
||||
import time as t
|
||||
|
||||
from rich.panel import Panel
|
||||
from rich.table import Table
|
||||
from rich.text import Text
|
||||
|
||||
from oraclevision.bip110 import BlockAnalysis, analyze_block
|
||||
from oraclevision.bitcoin_cli import BitcoinCLI, BitcoinCLIError
|
||||
from oraclevision.config import load_settings
|
||||
from oraclevision.mempool_compose import analyze_block_template
|
||||
from oraclevision.security import resolve_executable
|
||||
from oraclevision.spam_score import status_style
|
||||
from shared.display import clear
|
||||
from shared.rich_ui import console, rich_error, rich_prompt
|
||||
from shared.ui import show_error
|
||||
|
||||
|
||||
def _header() -> None:
|
||||
console.print()
|
||||
console.print(
|
||||
Panel(
|
||||
Text.from_markup(
|
||||
"[bold rgb(255,102,0)]OracleVision[/] · "
|
||||
"[dim]Don't Trust, Verify[/]\n"
|
||||
"[dim]Local BIP-110 & mempool analysis via bitcoin-cli[/]"
|
||||
),
|
||||
border_style="rgb(255,102,0)",
|
||||
)
|
||||
)
|
||||
console.print()
|
||||
|
||||
|
||||
def _menu_items() -> None:
|
||||
console.print(" [bold cyan]A.[/] BIP-110 Block Scanner")
|
||||
console.print(" [bold cyan]B.[/] Mempool Glass (getblocktemplate)")
|
||||
console.print(" [bold cyan]C.[/] Block Detail View")
|
||||
console.print(" [bold cyan]D.[/] Launch Full OracleVision TUI")
|
||||
console.print(" [bold yellow]R.[/] Return")
|
||||
console.print()
|
||||
|
||||
|
||||
def _cli_for(path: dict) -> BitcoinCLI:
|
||||
settings = load_settings()
|
||||
return BitcoinCLI.from_path_config(
|
||||
path,
|
||||
datadir=settings.bitcoin_datadir,
|
||||
timeout=float(settings.cli_timeout_seconds),
|
||||
)
|
||||
|
||||
|
||||
def _format_block_row(analysis: BlockAnalysis) -> tuple:
|
||||
sig = "Y" if analysis.bip110_signaling else "n"
|
||||
flags = []
|
||||
if analysis.violation_count:
|
||||
flags.append(f"bip110:{analysis.violation_count}")
|
||||
if analysis.inscription_count:
|
||||
flags.append(f"insc:{analysis.inscription_count}")
|
||||
if analysis.brc20_count:
|
||||
flags.append(f"brc20:{analysis.brc20_count}")
|
||||
if analysis.runes_count:
|
||||
flags.append(f"runes:{analysis.runes_count}")
|
||||
flag_text = ", ".join(flags) if flags else "—"
|
||||
return (
|
||||
str(analysis.height),
|
||||
analysis.miner_tag[:24],
|
||||
str(analysis.spam_score),
|
||||
f"[{status_style(analysis.status)}]{analysis.status}[/]",
|
||||
sig,
|
||||
flag_text,
|
||||
)
|
||||
|
||||
|
||||
def scan_recent_blocks(path: dict, count: int | None = None) -> None:
|
||||
"""Scan recent blocks for BIP-110 violations and spam signals."""
|
||||
settings = load_settings()
|
||||
count = count or settings.block_scan_count
|
||||
cli = _cli_for(path)
|
||||
|
||||
clear()
|
||||
_header()
|
||||
console.print(f"[dim]Scanning last {count} blocks from your node…[/]\n")
|
||||
|
||||
try:
|
||||
tip = cli.get_block_count()
|
||||
table = Table(title="BIP-110 Block Scanner", show_lines=True)
|
||||
table.add_column("Height", style="cyan", justify="right")
|
||||
table.add_column("Miner", style="white")
|
||||
table.add_column("Score", justify="right")
|
||||
table.add_column("Status")
|
||||
table.add_column("BIP110", justify="center")
|
||||
table.add_column("Flags", style="dim")
|
||||
|
||||
for height in range(tip, max(tip - count, -1), -1):
|
||||
block_hash = cli.get_block_hash(height)
|
||||
block = cli.get_block(block_hash, 2)
|
||||
analysis = analyze_block(block, spam_threshold=settings.spam_score_threshold)
|
||||
table.add_row(*_format_block_row(analysis))
|
||||
|
||||
console.print(table)
|
||||
console.print(
|
||||
"\n[dim]Score 0-100 · CLEAN / SUSPICIOUS / VIOLATION · "
|
||||
"BIP110 = version bit 4 signaling[/]"
|
||||
)
|
||||
except BitcoinCLIError as exc:
|
||||
rich_error(str(exc))
|
||||
if exc.hint:
|
||||
console.print(f" [dim]→ {exc.hint}[/]")
|
||||
except Exception as exc:
|
||||
show_error(str(exc))
|
||||
|
||||
input("\n\aContinue...")
|
||||
|
||||
|
||||
def show_mempool_glass(path: dict) -> None:
|
||||
"""Show Mempool Glass composition from getblocktemplate."""
|
||||
cli = _cli_for(path)
|
||||
|
||||
clear()
|
||||
_header()
|
||||
console.print("[dim]Fetching block template from your node…[/]\n")
|
||||
|
||||
try:
|
||||
template = cli.get_block_template()
|
||||
composition = analyze_block_template(template, cli.decode_raw_transaction)
|
||||
mempool = cli.get_mempool_info()
|
||||
|
||||
if composition.error:
|
||||
rich_error(composition.error)
|
||||
input("\n\aContinue...")
|
||||
return
|
||||
|
||||
summary = Table(title="Mempool Glass — Block Template", show_header=False)
|
||||
summary.add_column("Metric", style="cyan")
|
||||
summary.add_column("Value", style="white")
|
||||
summary.add_row("Template height", str(composition.template_height))
|
||||
summary.add_row("Mempool txs", str(mempool.get("size", "?")))
|
||||
summary.add_row("Template txs", str(composition.total_tx))
|
||||
summary.add_row("Analyzed txs", str(composition.analyzed_tx))
|
||||
summary.add_row("Template weight", f"{composition.analyzed_weight:,} / {composition.weight_limit:,}")
|
||||
summary.add_row("Fill", f"{composition.fill_pct:.1f}%")
|
||||
|
||||
cats = Table(title="Transaction Categories", show_lines=True)
|
||||
cats.add_column("Category", style="bold")
|
||||
cats.add_column("Count", justify="right")
|
||||
cats.add_column("Weight", justify="right")
|
||||
cats.add_column("% of template", justify="right")
|
||||
|
||||
rows = [
|
||||
("economic", composition.economic_count, composition.economic_weight, "green"),
|
||||
("consolidation", composition.consolidation_count, composition.consolidation_weight, "cyan"),
|
||||
("coinjoin", composition.coinjoin_count, composition.coinjoin_weight, "blue"),
|
||||
("spam", composition.spam_count, composition.spam_weight, "red"),
|
||||
]
|
||||
for name, cnt, wt, color in rows:
|
||||
cats.add_row(
|
||||
f"[{color}]{name}[/]",
|
||||
str(cnt),
|
||||
f"{wt:,}",
|
||||
f"{composition.pct(wt):.1f}%",
|
||||
)
|
||||
|
||||
console.print(summary)
|
||||
console.print()
|
||||
console.print(cats)
|
||||
console.print(
|
||||
"\n[dim]Spam = BIP-110 violations, inscriptions, tokens, "
|
||||
"oversized witness, or excess witness ratio[/]"
|
||||
)
|
||||
except BitcoinCLIError as exc:
|
||||
rich_error(str(exc))
|
||||
if exc.hint:
|
||||
console.print(f" [dim]→ {exc.hint}[/]")
|
||||
except Exception as exc:
|
||||
show_error(str(exc))
|
||||
|
||||
input("\n\aContinue...")
|
||||
|
||||
|
||||
def _render_block_detail(analysis: BlockAnalysis) -> None:
|
||||
sig = "YES" if analysis.bip110_signaling else "no"
|
||||
title = (
|
||||
f"Block #{analysis.height} · Spam {analysis.spam_score}/100 · "
|
||||
f"{analysis.status} · BIP110 bit4: {sig}"
|
||||
)
|
||||
|
||||
info = Table(show_header=False, title=title)
|
||||
info.add_column("Field", style="cyan")
|
||||
info.add_column("Value")
|
||||
info.add_row("Hash", analysis.hash)
|
||||
info.add_row("Miner", analysis.miner_tag)
|
||||
info.add_row("Weight", f"{analysis.weight:,} ({analysis.tx_count} txs)")
|
||||
info.add_row("Witness", f"{analysis.witness_pct:.1f}% of block weight")
|
||||
info.add_row("Inscriptions", str(analysis.inscription_count))
|
||||
info.add_row("BRC-20", str(analysis.brc20_count))
|
||||
info.add_row("Runes", str(analysis.runes_count))
|
||||
info.add_row("OP_RETURN", str(analysis.op_return_count))
|
||||
info.add_row(
|
||||
"BIP-110 violations",
|
||||
f"{analysis.violation_count} txs ({analysis.violation_weight:,} wt)",
|
||||
)
|
||||
|
||||
console.print(info)
|
||||
console.print()
|
||||
|
||||
bad = [tx for tx in analysis.transactions if tx.has_bip110_violation or tx.is_spam_signal]
|
||||
bad.sort(key=lambda tx: tx.weight, reverse=True)
|
||||
|
||||
if not bad:
|
||||
console.print("[green]No problematic transactions detected.[/]")
|
||||
return
|
||||
|
||||
tx_table = Table(title="Problematic Transactions (top 25)", show_lines=True)
|
||||
tx_table.add_column("TXID", style="red")
|
||||
tx_table.add_column("Weight", justify="right")
|
||||
tx_table.add_column("BIP-110 flags")
|
||||
tx_table.add_column("Signals")
|
||||
|
||||
for tx in bad[:25]:
|
||||
flags = ", ".join(sorted(tx.bip110_flags)) or "—"
|
||||
signals = ", ".join(sorted(tx.signals)) or "—"
|
||||
tx_table.add_row(tx.txid[:20] + "…", f"{tx.weight:,}", flags, signals)
|
||||
|
||||
console.print(tx_table)
|
||||
if len(bad) > 25:
|
||||
console.print(f"[dim]… and {len(bad) - 25} more[/]")
|
||||
|
||||
|
||||
def show_block_detail(path: dict, target: str | None = None) -> None:
|
||||
"""Analyze a single block by height or hash."""
|
||||
cli = _cli_for(path)
|
||||
settings = load_settings()
|
||||
|
||||
clear()
|
||||
_header()
|
||||
|
||||
if not target:
|
||||
target = input("\033[1;32;40mBlock height or hash: \033[0;37;40m").strip()
|
||||
if not target:
|
||||
return
|
||||
|
||||
try:
|
||||
if target.isdigit():
|
||||
block_hash = cli.get_block_hash(int(target))
|
||||
else:
|
||||
block_hash = target
|
||||
|
||||
console.print(f"[dim]Loading block {block_hash[:16]}…[/]\n")
|
||||
block = cli.get_block(block_hash, 2)
|
||||
analysis = analyze_block(block, spam_threshold=settings.spam_score_threshold)
|
||||
_render_block_detail(analysis)
|
||||
except BitcoinCLIError as exc:
|
||||
rich_error(str(exc))
|
||||
if exc.hint:
|
||||
console.print(f" [dim]→ {exc.hint}[/]")
|
||||
except Exception as exc:
|
||||
show_error(str(exc))
|
||||
|
||||
input("\n\aContinue...")
|
||||
|
||||
|
||||
def launch_full_oraculovision(path: dict) -> None:
|
||||
"""Launch the standalone OracleVision Textual TUI if installed."""
|
||||
settings = load_settings()
|
||||
command = settings.oraculovision_command
|
||||
|
||||
clear()
|
||||
_header()
|
||||
|
||||
try:
|
||||
launch_cmd = resolve_executable(command)
|
||||
except ValueError as exc:
|
||||
rich_error(str(exc))
|
||||
console.print(
|
||||
" [dim]→ Install OracleVision: pip install -e . from "
|
||||
"https://github.com/MarcanoFilms/oraculovision[/]"
|
||||
)
|
||||
input("\n\aContinue...")
|
||||
return
|
||||
|
||||
console.print(f"[dim]Launching {launch_cmd[0]}…[/]\n")
|
||||
console.print("[yellow]Press Ctrl+C in OracleVision to return to PyBLOCK.[/]\n")
|
||||
t.sleep(1)
|
||||
|
||||
env = dict(**{k: v for k, v in __import__("os").environ.items()})
|
||||
if settings.bitcoin_datadir:
|
||||
env["BITCOIN_DATADIR"] = settings.bitcoin_datadir
|
||||
if path.get("bitcoincli"):
|
||||
env["BITCOIN_CLI"] = path["bitcoincli"]
|
||||
|
||||
try:
|
||||
# nosemgrep: python.lang.security.audit.dangerous-subprocess-use-audit
|
||||
subprocess.run(launch_cmd, env=env, check=False)
|
||||
except FileNotFoundError:
|
||||
rich_error(f"Could not execute: {launch_cmd[0]}")
|
||||
except KeyboardInterrupt:
|
||||
pass
|
||||
|
||||
input("\n\aContinue...")
|
||||
|
||||
|
||||
def run_oraclevision_menu(path: dict) -> None:
|
||||
"""Main OracleVision submenu loop."""
|
||||
settings = load_settings()
|
||||
if settings.load_error:
|
||||
rich_error(f"Config warning: {settings.load_error}")
|
||||
|
||||
while True:
|
||||
clear()
|
||||
_header()
|
||||
_menu_items()
|
||||
choice = rich_prompt("Select option").strip().upper()
|
||||
|
||||
if choice in ("A",):
|
||||
scan_recent_blocks(path)
|
||||
elif choice in ("B",):
|
||||
show_mempool_glass(path)
|
||||
elif choice in ("C",):
|
||||
show_block_detail(path)
|
||||
elif choice in ("D",):
|
||||
launch_full_oraculovision(path)
|
||||
elif choice in ("R", ""):
|
||||
break
|
||||
else:
|
||||
show_error(f"Invalid option '{choice}'")
|
||||
t.sleep(1)
|
||||
Loading…
Add table
Add a link
Reference in a new issue