feat: machine-checkable custody policy enforcement

.github/workflows/ci.yml

@@ -23,3 +23,9 @@ jobs:
 
  - name: Run tests
  run: python -m pytest -q
+
+ - name: Validate custody policy
+ run: python -c "from services.custody_policy.policy import load_policy; load_policy(); print('custody policy: valid')"
+
+ - name: Run custody policy tests
+ run: python -m pytest -q tests/custody_policy

requirements-dev.txt

@@ -1,3 +1,6 @@
 pytest>=8.4,<9.0
 httpx>=0.28,<1.0
+pyyaml>=6.0,<7.0
+jsonschema>=4.25,<5.0
+web3>=7.13,<8.0
 -r services/api/requirements.txt

services/custody_policy/README.md

@@ -0,0 +1,14 @@
+# Custody policy package
+
+This package provides deterministic, fail-closed custody validation.
+
+## Components
+
+- `custody-policy.yaml`: policy configuration.
+- `custody-policy.schema.json`: JSON Schema validation.
+- `scoring.py`: deterministic signer and transaction risk scoring.
+- `collector.py`: read-only Safe state collection over JSON-RPC.
+- `policy.py`: policy loading and schema validation.
+- `cli.py`: local fixture or live RPC evaluation.
+
+The collector is read-only and never accepts private keys or signing material.

services/custody_policy/__init__.py

@@ -0,0 +1,4 @@
+from .policy import evaluate, load_policy
+from .scoring import evaluate_custody, signer_independence, transaction_risk
+
+__all__ = ["evaluate", "evaluate_custody", "load_policy", "signer_independence", "transaction_risk"]

services/custody_policy/cli.py

@@ -0,0 +1,35 @@
+from __future__ import annotations
+
+import argparse
+import json
+
+from .collector import collect_safe_state
+from .policy import evaluate
+
+
+def main() -> int:
+ parser = argparse.ArgumentParser(description="SentinelAI deterministic custody policy validator")
+ parser.add_argument("--state", help="JSON state fixture to evaluate")
+ parser.add_argument("--rpc-url", help="JSON-RPC URL for live Safe collection")
+ parser.add_argument("--safe", help="Safe address")
+ parser.add_argument("--chain-id", type=int)
+ args = parser.parse_args()
+
+ if bool(args.state) == bool(args.rpc_url):
+ parser.error("provide exactly one of --state or --rpc-url")
+
+ if args.state:
+ with open(args.state, "r", encoding="utf-8") as handle:
+ state = json.load(handle)
+ else:
+ if not args.safe:
+ parser.error("--safe is required with --rpc-url")
+ state = collect_safe_state(args.rpc_url, args.safe, args.chain_id).as_dict()
+
+ result = evaluate(state)
+ print(json.dumps(result, indent=2, sort_keys=True))
+ return 0 if result["decision"] != "BLOCK" else 2
+
+
+if __name__ == "__main__":
+ raise SystemExit(main())

services/custody_policy/collector.py

@@ -0,0 +1,111 @@
+from __future__ import annotations
+
+from dataclasses import asdict, dataclass
+from typing import Any
+
+from web3 import Web3
+
+FALLBACK_HANDLER_STORAGE_SLOT = "0x6c9a6c4a39284e37ed1cf53d337577d14212a4870fb976a4366c693b939918d5"
+GUARD_STORAGE_SLOT = "0x4a204f620c8c5ccdca3fd54d003badd85ba500436a431f0cbda4f558c93c34c8"
+MODULE_GUARD_STORAGE_SLOT = "0xb104e0b93118902c651344349b610029d694cfdec91c589c91ebafbcd0289947"
+SAFE_ABI = [
+ {"inputs": [], "name": "getOwners", "outputs": [{"internalType": "address[]", "name": "", "type": "address[]"}], "stateMutability": "view", "type": "function"},
+ {"inputs": [], "name": "getThreshold", "outputs": [{"internalType": "uint256", "name": "", "type": "uint256"}], "stateMutability": "view", "type": "function"},
+ {"inputs": [{"internalType": "address", "name": "start", "type": "address"}, {"internalType": "uint256", "name": "pageSize", "type": "uint256"}], "name": "getModulesPaginated", "outputs": [{"internalType": "address[]", "name": "array", "type": "address[]"}, {"internalType": "address", "name": "next", "type": "address"}], "stateMutability": "view", "type": "function"},
+ {"inputs": [], "name": "VERSION", "outputs": [{"internalType": "string", "name": "", "type": "string"}], "stateMutability": "view", "type": "function"},
+]
+ZERO = "0x0000000000000000000000000000000000000000"
+SENTINEL = "0x0000000000000000000000000000000000000001"
+
+
+@dataclass(frozen=True)
+class SafeState:
+ chain_id: int
+ address: str
+ owners: list[str]
+ threshold: int
+ modules: list[dict[str, str]]
+ guards: list[dict[str, str]]
+ module_guard: dict[str, str] | None
+ fallback_handler: dict[str, str] | None
+ version: str | None
+ signers: list[dict[str, Any]]
+
+ def as_dict(self) -> dict[str, Any]:
+ return asdict(self)
+
+
+def _address_from_storage(value: bytes | str) -> str:
+ raw = value.hex() if isinstance(value, bytes) else value.removeprefix("0x")
+ return Web3.to_checksum_address("0x" + raw[-40:])
+
+
+def _code_hash(w3: Web3, address: str) -> str:
+ code = w3.eth.get_code(address)
+ if not code:
+ return ZERO
+ return Web3.to_hex(Web3.keccak(code))
+
+
+def _extension(w3: Web3, address: str) -> dict[str, str]:
+ address = Web3.to_checksum_address(address)
+ return {"address": address, "code_hash": _code_hash(w3, address)}
+
+
+def _walk_modules(contract: Any, page_size: int = 100) -> list[str]:
+ modules: list[str] = []
+ cursor = SENTINEL
+ for _ in range(100):
+ page, next_cursor = contract.functions.getModulesPaginated(cursor, page_size).call()
+ modules.extend(Web3.to_checksum_address(x) for x in page)
+ if not page or Web3.to_checksum_address(next_cursor) == Web3.to_checksum_address(SENTINEL):
+ break
+ cursor = next_cursor
+ else:
+ raise RuntimeError("module pagination exceeded safety limit")
+ return modules
+
+
+def collect_safe_state(rpc_url: str, safe_address: str, chain_id: int | None = None) -> SafeState:
+ w3 = Web3(Web3.HTTPProvider(rpc_url, request_kwargs={"timeout": 20}))
+ if not w3.is_connected():
+ raise ConnectionError("unable to connect to JSON-RPC endpoint")
+ actual_chain_id = int(w3.eth.chain_id)
+ if chain_id is not None and actual_chain_id != chain_id:
+ raise ValueError(f"chain ID mismatch: expected {chain_id}, got {actual_chain_id}")
+
+ address = Web3.to_checksum_address(safe_address)
+ if w3.eth.get_code(address) in (b"", b"\x00"):
+ raise ValueError("target address has no deployed bytecode")
+ contract = w3.eth.contract(address=address, abi=SAFE_ABI)
+ owners = [Web3.to_checksum_address(x) for x in contract.functions.getOwners().call()]
+ threshold = int(contract.functions.getThreshold().call())
+ modules = [_extension(w3, x) for x in _walk_modules(contract)]
+
+ fallback_address = _address_from_storage(w3.eth.get_storage_at(address, FALLBACK_HANDLER_STORAGE_SLOT))
+ guard_address = _address_from_storage(w3.eth.get_storage_at(address, GUARD_STORAGE_SLOT))
+ module_guard_address = _address_from_storage(w3.eth.get_storage_at(address, MODULE_GUARD_STORAGE_SLOT))
+ guards = [] if guard_address == Web3.to_checksum_address(ZERO) else [_extension(w3, guard_address)]
+ module_guard = None if module_guard_address == Web3.to_checksum_address(ZERO) else _extension(w3, module_guard_address)
+ fallback = None if fallback_address == Web3.to_checksum_address(ZERO) else _extension(w3, fallback_address)
+
+ try:
+ version = str(contract.functions.VERSION().call())
+ except Exception:
+ version = None
+
+ # On-chain state cannot establish operational independence. Unknown metadata is
+ # intentionally false so the policy engine fails closed until an operator enriches it.
+ signers = [{
+ "address": owner,
+ "key_generation_verified": False,
+ "hardware_independent": False,
+ "software_independent": False,
+ "administrator_independent": False,
+ "geography_independent": False,
+ "backup_independent": False,
+ "communications_independent": False,
+ "recovery_independent": False,
+ } for owner in owners]
+
+ return SafeState(actual_chain_id, address, owners, threshold, modules, guards, module_guard, fallback, version, signers)

services/custody_policy/custody-policy.schema.json

@@ -0,0 +1,53 @@
+{
+ "$schema": "https://json-schema.org/draft/2020-12/schema",
+ "$id": "https://sentinelai.dev/schemas/custody-policy.schema.json",
+ "title": "SentinelAI Custody Policy",
+ "type": "object",
+ "required": ["version", "mode", "custody", "modules", "guards", "allowlists", "transaction", "recovery", "risk", "hard_blocks"],
+ "properties": {
+ "version": {"type": "string", "pattern": "^[0-9]+\\.[0-9]+$"},
+ "mode": {"const": "FAIL_CLOSED"},
+ "custody": {
+ "type": "object", "required": ["minimum_threshold", "require_threshold_below_signer_count", "minimum_signer_independence", "preferred_signer_independence", "minimum_recovery_score"],
+ "properties": {
+ "minimum_threshold": {"type": "integer", "minimum": 2},
+ "require_threshold_below_signer_count": {"const": true},
+ "minimum_signer_independence": {"type": "integer", "minimum": 0, "maximum": 100},
+ "preferred_signer_independence": {"type": "integer", "minimum": 0, "maximum": 100},
+ "minimum_recovery_score": {"type": "integer", "minimum": 0, "maximum": 100}
+ }, "additionalProperties": false
+ },
+ "modules": {"$ref": "#/$defs/extensionPolicy"},
+ "guards": {"$ref": "#/$defs/extensionPolicyWithRecovery"},
+ "allowlists": {
+ "type": "object", "additionalProperties": false,
+ "required": ["modules", "guards", "module_guards", "fallback_handlers"],
+ "properties": {
+ "modules": {"$ref": "#/$defs/allowlist"},
+ "guards": {"$ref": "#/$defs/allowlist"},
+ "module_guards": {"$ref": "#/$defs/allowlist"},
+ "fallback_handlers": {"$ref": "#/$defs/allowlist"}
+ }
+ },
+ "transaction": {
+ "type": "object", "additionalProperties": false,
+ "required": ["require_destination_verification", "require_calldata_decoding", "require_simulation_for_high_value", "block_unlimited_approval", "block_unknown_delegatecall"],
+ "properties": {
+ "require_destination_verification": {"const": true},
+ "require_calldata_decoding": {"const": true},
+ "require_simulation_for_high_value": {"const": true},
+ "block_unlimited_approval": {"const": true},
+ "block_unknown_delegatecall": {"const": true}
+ }
+ },
+ "recovery": {"type": "object", "additionalProperties": false, "required": ["require_test", "max_age_days"], "properties": {"require_test": {"const": true}, "max_age_days": {"type": "integer", "minimum": 1}}},
+ "risk": {"type": "object", "required": ["bands"], "properties": {"bands": {"type": "object", "required": ["low_max", "moderate_max", "high_max", "critical_max"], "properties": {"low_max": {"const": 19}, "moderate_max": {"const": 39}, "high_max": {"const": 59}, "critical_max": {"const": 79}}, "additionalProperties": false}}, "additionalProperties": false},
+ "hard_blocks": {"type": "array", "items": {"type": "string"}, "uniqueItems": true}
+ },
+ "$defs": {
+ "extensionPolicy": {"type": "object", "additionalProperties": false, "required": ["default", "require_source_verified", "require_code_hash", "require_audit_for_material"], "properties": {"default": {"const": "DENY"}, "require_source_verified": {"const": true}, "require_code_hash": {"const": true}, "require_audit_for_material": {"const": true}}},
+ "extensionPolicyWithRecovery": {"type": "object", "additionalProperties": false, "required": ["default", "require_source_verified", "require_code_hash", "require_audit_for_material", "require_recovery_test"], "properties": {"default": {"const": "DENY"}, "require_source_verified": {"const": true}, "require_code_hash": {"const": true}, "require_audit_for_material": {"const": true}, "require_recovery_test": {"const": true}}},
+ "allowlist": {"type": "array", "items": {"type": "object", "required": ["chain_id", "address", "code_hash"], "properties": {"chain_id": {"type": "integer", "minimum": 1}, "address": {"type": "string", "pattern": "^0x[0-9a-fA-F]{40}$"}, "code_hash": {"type": "string", "pattern": "^0x[0-9a-fA-F]{64}$"}, "audit_status": {"type": "string"}, "max_value": {"type": "string"}}, "additionalProperties": false}, "uniqueItems": true}
+ },
+ "additionalProperties": false
+}

services/custody_policy/custody-policy.yaml

@@ -0,0 +1,49 @@
+version: "1.0"
+mode: FAIL_CLOSED
+custody:
+ minimum_threshold: 2
+ require_threshold_below_signer_count: true
+ minimum_signer_independence: 70
+ preferred_signer_independence: 85
+ minimum_recovery_score: 80
+modules:
+ default: DENY
+ require_source_verified: true
+ require_code_hash: true
+ require_audit_for_material: true
+guards:
+ default: DENY
+ require_source_verified: true
+ require_code_hash: true
+ require_audit_for_material: true
+ require_recovery_test: true
+allowlists:
+ modules: []
+ guards: []
+ module_guards: []
+ fallback_handlers: []
+transaction:
+ require_destination_verification: true
+ require_calldata_decoding: true
+ require_simulation_for_high_value: true
+ block_unlimited_approval: true
+ block_unknown_delegatecall: true
+recovery:
+ require_test: true
+ max_age_days: 365
+risk:
+ bands:
+ low_max: 19
+ moderate_max: 39
+ high_max: 59
+ critical_max: 79
+hard_blocks:
+ - unknown_module
+ - unknown_guard
+ - unknown_module_guard
+ - unknown_fallback
+ - simulation_mismatch
+ - unauthorized_owner_change
+ - unauthorized_threshold_change
+ - unbounded_delegatecall
+ - denylisted_destination

services/custody_policy/policy.py

@@ -0,0 +1,29 @@
+from __future__ import annotations
+
+import json
+from pathlib import Path
+from typing import Any
+
+import yaml
+from jsonschema import Draft202012Validator
+
+from .scoring import evaluate_custody
+
+ROOT = Path(__file__).resolve().parent
+SCHEMA_PATH = ROOT / "custody-policy.schema.json"
+POLICY_PATH = ROOT / "custody-policy.yaml"
+
+
+def load_policy(path: str | Path = POLICY_PATH) -> dict[str, Any]:
+ with Path(path).open("r", encoding="utf-8") as handle:
+ policy = yaml.safe_load(handle)
+ schema = json.loads(SCHEMA_PATH.read_text(encoding="utf-8"))
+ errors = sorted(Draft202012Validator(schema).iter_errors(policy), key=lambda e: list(e.path))
+ if errors:
+ details = "; ".join(f"{'.'.join(map(str, e.path))}: {e.message}" for e in errors)
+ raise ValueError(f"invalid custody policy: {details}")
+ return policy
+
+
+def evaluate(state: dict[str, Any], policy: dict[str, Any] | None = None) -> dict[str, Any]:
+ return evaluate_custody(state, policy or load_policy())