Transaction Policy Engine Development

Development of Transaction Policy Engine

Classic scenario: custodial wallet or corporate multisig. An employee initiates $500K withdrawal to external address. The contract has no objections — signatures collected. But the address fell under OFAC sanctions two hours ago. Or daily withdrawal limit exceeded because three other transactions went through API simultaneously. Or it's just an address with risk score 95 on Chainalysis.

Policy engine is a layer between transaction initiator and execution that applies set of rules before signing. Not after, not through monitoring — exactly before. At this level you can stop transaction, require additional signatures, log for compliance, or auto-reject.

Policy Engine Architecture

Levels of Policy Application

Policy engine can exist at several levels — they're often mixed, creating problems:

Off-chain pre-execution — most common and practical. Rules checked in service before transaction goes to signature and mempool. Flexible, cheap, supports any logic. Disadvantage: requires trust in this service; if compromised — policies bypassed.

On-chain enforcement — smart contract that's entry point for all transactions and contains policy logic. Safe{Core} Protocol Hooks example of such mechanism. Stronger guarantees, but logic limited to what can be expressed on-chain: no external data access without oracles, each check costs gas.

Hybrid — policies verified off-chain, but contract accepts proof that check passed (commitment scheme or trusted signer attestation).

Rule Model

Rule consists of condition and action. Conditions can be:

• Parametric: amount > threshold, recipient in whitelist, token == USDC
• Contextual: sender.role == OPERATOR, time_of_day in 09:00-18:00, daily_volume + amount <= limit
• External: chainalysis_risk_score(recipient) < 70, ofac_check(recipient) == CLEAR
• On-chain: recipient.is_contract == false, token.paused == false

Actions: ALLOW, DENY, REQUIRE_APPROVAL(n_signers), DELAY(duration), NOTIFY(channels).

Rules have priorities and can conflict. Need clear semantics: first-match, or all-must-pass, or whitelist-overrides-blacklist. This is design decision baked into engine.

interface PolicyRule {
  id: string;
  priority: number;
  conditions: Condition[];
  conditionLogic: 'AND' | 'OR';
  action: PolicyAction;
  metadata: { name: string; owner: string; updatedAt: number };
}

interface PolicyAction {
  type: 'ALLOW' | 'DENY' | 'REQUIRE_APPROVAL' | 'DELAY';
  params?: {
    requiredApprovers?: string[];    // addresses or roles
    minApprovals?: number;
    delaySeconds?: number;
    notifyChannels?: string[];
  };
}

Evaluator: Computation Order

Engine must process rules efficiently — especially when hundreds of rules, and some require external calls (compliance provider API).

Optimal strategy: short-circuit evaluation with caching. Check cheap local conditions first (transaction parameters, roles), then — cached external data, last — fresh API calls with timeout.

class PolicyEvaluator:
    def evaluate(self, tx: Transaction, context: EvalContext) -> PolicyDecision:
        sorted_rules = sorted(self.rules, key=lambda r: r.priority, reverse=True)
        
        for rule in sorted_rules:
            # Check cheap conditions first
            cheap_conditions = [c for c in rule.conditions if c.type == 'PARAMETRIC']
            if not self._eval_conditions(cheap_conditions, tx, context):
                continue
            
            # Then expensive (with cache)
            expensive_conditions = [c for c in rule.conditions if c.type == 'EXTERNAL']
            cache_key = self._cache_key(expensive_conditions, tx)
            cached = self.cache.get(cache_key)
            
            results = cached if cached else self._eval_external(expensive_conditions, tx)
            self.cache.set(cache_key, results, ttl=300)
            
            if self._eval_conditions_with_results(rule.conditions, results, rule.conditionLogic):
                return PolicyDecision(action=rule.action, rule_id=rule.id)
        
        return PolicyDecision(action=DEFAULT_ACTION)

On-Chain Implementation: Safe Hooks

Safe{Core} Protocol (EIP-7579 compatible) provides hooks mechanism:

interface ISafeProtocolHooks {
    function preCheck(
        Safe safe,
        SafeTransaction calldata tx,
        uint256 executionType,
        bytes calldata executionMeta
    ) external returns (bytes memory preCheckData);
    
    function postCheck(
        Safe safe,
        bool success,
        bytes calldata preCheckData
    ) external;
}

preCheck called before execution. If reverts — transaction doesn't pass. Here you can: check whitelist/blacklist (stored in hook storage), check limits (via accumulators per address/token), require additional approval via timelock.

Example limit hook:

contract DailyLimitHook is ISafeProtocolHooks {
    mapping(address => mapping(address => uint256)) public dailyVolume; // safe => token => amount
    mapping(address => mapping(address => uint256)) public lastResetDay;
    mapping(address => mapping(address => uint256)) public dailyLimit;

    function preCheck(Safe safe, SafeTransaction calldata tx, uint256, bytes calldata)
        external returns (bytes memory)
    {
        address token = _extractToken(tx.data);
        uint256 amount = _extractAmount(tx.data);
        
        uint256 today = block.timestamp / 1 days;
        address safeAddr = address(safe);
        
        if (lastResetDay[safeAddr][token] < today) {
            dailyVolume[safeAddr][token] = 0;
            lastResetDay[safeAddr][token] = today;
        }
        
        require(
            dailyVolume[safeAddr][token] + amount <= dailyLimit[safeAddr][token],
            "DailyLimitExceeded"
        );
        
        return abi.encode(token, amount); // pass to postCheck
    }

    function postCheck(Safe safe, bool success, bytes calldata preCheckData) external {
        if (success) {
            (address token, uint256 amount) = abi.decode(preCheckData, (address, uint256));
            dailyVolume[address(safe)][token] += amount;
        }
    }
}

Compliance Integrations

For financial products, policy engine inevitably includes compliance provider integration. Main ones:

Chainalysis — KYT API. Address checking (risk score) and transactions (exposure to known clusters). Latency: 200–800ms, need cache and graceful degradation.

Elliptic — similar functionality, slightly different risk assessment model. Used in Fireblocks.

TRM Labs — specializes in cross-chain analysis, good coverage of Solana and Tron.

OFAC screening — can go through same providers or self-maintained SDN list snapshot (updates infrequently, can store locally and update via webhook).

Important: compliance API have SLA and can be unavailable. Policy engine should have explicit policy for EXTERNAL_CHECK_TIMEOUT: fail-open (allow with log) vs. fail-closed (block). This is business decision but must be fixed.

Monitoring and Audit

Policy engine without complete audit log is useless for compliance. Each decision should contain:

• Transaction hash or pre-tx ID
• List of applied rules and results
• Values of all conditions at evaluation time
• Final decision and executor
• Timestamp with millisecond precision

This is immutable log. Storage: PostgreSQL with append-only table + replication to S3/Arweave for long-term retention. For regulatory requirements — minimum 5 years.

Developing policy engine is not just if-else statements. It's system with formal rule semantics, atomicity guarantees, auditability and operational reliability. Done right, it becomes foundation for obtaining licenses and passing compliance audits.