foundry_evm_symbolic/executor/
invariant.rs

1use super::*;
2
3impl SymbolicExecutor {
4    #[expect(clippy::too_many_arguments)]
5    pub(super) fn execute_invariant_check<FEN: FoundryEvmNetwork>(
6        &mut self,
7        executor: &Executor<FEN>,
8        state: PathState,
9        invariant_address: Address,
10        sender: Address,
11        invariant: &Function,
12        after_invariant: Option<&Function>,
13        completed_paths: &mut usize,
14    ) -> Result<Vec<InvariantCheckOutcome>, SymbolicError> {
15        let calldata = SymbolicCalldata::selector_only(&mut self.cx, invariant)?;
16        let call_data = calldata.call_data(&mut self.cx);
17        let constraints = calldata.into_constraints();
18        let outcomes = self.execute_sequence_call(
19            executor,
20            state,
21            invariant_address,
22            sender,
23            invariant,
24            call_data,
25            constraints,
26            completed_paths,
27        )?;
28
29        let mut checked = Vec::new();
30        for mut outcome in outcomes {
31            if !matches!(outcome.status, TopLevelCallStatus::Success) {
32                outcome.status = TopLevelCallStatus::Failure;
33                checked.push(InvariantCheckOutcome { failed: true, state: outcome.state });
34                continue;
35            }
36
37            if self.invariant_return_failed(invariant, &outcome.return_data, &mut outcome.state)? {
38                checked.push(InvariantCheckOutcome { failed: true, state: outcome.state });
39                continue;
40            }
41
42            let Some(after_invariant) = after_invariant else {
43                checked.push(InvariantCheckOutcome { failed: false, state: outcome.state });
44                continue;
45            };
46
47            let after_calldata = SymbolicCalldata::selector_only(&mut self.cx, after_invariant)?;
48            let calldata = after_calldata.call_data(&mut self.cx);
49            let constraints = after_calldata.constraints().to_vec();
50            for after_outcome in self.execute_sequence_call(
51                executor,
52                outcome.state.clone(),
53                invariant_address,
54                sender,
55                after_invariant,
56                calldata,
57                constraints,
58                completed_paths,
59            )? {
60                checked.push(InvariantCheckOutcome {
61                    failed: !matches!(after_outcome.status, TopLevelCallStatus::Success),
62                    state: after_outcome.state,
63                });
64            }
65        }
66        Ok(checked)
67    }
68
69    pub(super) fn invariant_return_failed(
70        &mut self,
71        invariant: &Function,
72        return_data: &SymReturnData,
73        state: &mut PathState,
74    ) -> Result<bool, SymbolicError> {
75        if invariant.outputs.is_empty() {
76            return Ok(false);
77        }
78        if invariant.outputs.len() != 1 || invariant.outputs[0].selector_type().as_ref() != "bool" {
79            return Ok(false);
80        }
81        if return_data.len() < 32 {
82            return Ok(true);
83        }
84
85        let pass = return_data.load_word(&mut self.cx, 0)?.nonzero_bool(&mut self.cx);
86        let fail = pass.clone().not(&mut self.cx);
87        match fail.as_const() {
88            Some(true) => Ok(true),
89            Some(false) => Ok(false),
90            None => {
91                let mut constraints = state.constraints.clone();
92                constraints.push(fail);
93                if self.solver.is_sat(&mut self.cx, &constraints)? {
94                    state.constraints = constraints;
95                    Ok(true)
96                } else {
97                    state.constraints.push(pass);
98                    Ok(false)
99                }
100            }
101        }
102    }
103
104    #[expect(clippy::too_many_arguments)]
105    pub(super) fn execute_sequence_call<FEN: FoundryEvmNetwork>(
106        &mut self,
107        executor: &Executor<FEN>,
108        mut state: PathState,
109        target: Address,
110        sender: Address,
111        _function: &Function,
112        calldata: SymCalldata,
113        constraints: Vec<SymBoolExpr>,
114        completed_paths: &mut usize,
115    ) -> Result<Vec<TopLevelCallOutcome>, SymbolicError> {
116        state.world.clear_transaction_scoped_state();
117        let code = state.world.extcode(&mut self.cx, executor, target)?;
118        state.call_depth = 0;
119        state.origin = sender;
120        state.origin_word = SymExpr::constant(&mut self.cx, address_word(sender));
121        let callvalue = SymExpr::zero(&mut self.cx);
122        state.frame = CallFrame::new(
123            &mut self.cx,
124            target,
125            target,
126            target,
127            sender,
128            callvalue,
129            false,
130            calldata,
131        );
132        state.constraints.extend(constraints);
133
134        let mut worklist = VecDeque::from([state]);
135        let mut outcomes = Vec::new();
136        let path_limit = self.config.path_width() as usize;
137        let depth_limit = self.config.execution_depth() as usize;
138
139        while let Some(mut state) = self.pop_next_feasible_path(&mut worklist)? {
140            if *completed_paths >= path_limit {
141                return Err(SymbolicError::Unsupported("symbolic path limit exceeded"));
142            }
143            let _path_span =
144                trace_span!("symbolic_path", completed_paths, worklist_size = worklist.len())
145                    .entered();
146            trace!(completed_paths, worklist_size = worklist.len(), "exploring symbolic path");
147
148            loop {
149                self.check_timeout()?;
150                if state.depth >= depth_limit {
151                    return Err(SymbolicError::Unsupported("symbolic depth limit exceeded"));
152                }
153                state.depth += 1;
154
155                let Some(op) = code.opcode(&mut self.cx, state.pc)? else {
156                    *completed_paths += 1;
157                    outcomes.push(TopLevelCallOutcome {
158                        status: if state.expectations_satisfied() {
159                            TopLevelCallStatus::Success
160                        } else {
161                            TopLevelCallStatus::Failure
162                        },
163                        return_data: state.return_data.clone(),
164                        state,
165                    });
166                    break;
167                };
168
169                let _step_span = trace_span!("symbolic_step", pc = state.pc, op).entered();
170                match self.step(
171                    executor,
172                    &code,
173                    code.jump_table(),
174                    &mut state,
175                    &mut worklist,
176                    completed_paths,
177                    op,
178                )? {
179                    StepOutcome::Continue => {}
180                    StepOutcome::Halt => {
181                        *completed_paths += 1;
182                        outcomes.push(TopLevelCallOutcome {
183                            status: if state.expectations_satisfied() {
184                                TopLevelCallStatus::Success
185                            } else {
186                                TopLevelCallStatus::Failure
187                            },
188                            return_data: state.return_data.clone(),
189                            state,
190                        });
191                        break;
192                    }
193                    StepOutcome::Revert => {
194                        *completed_paths += 1;
195                        outcomes.push(TopLevelCallOutcome {
196                            status: TopLevelCallStatus::Revert,
197                            return_data: state.return_data.clone(),
198                            state,
199                        });
200                        break;
201                    }
202                    StepOutcome::Failure => {
203                        *completed_paths += 1;
204                        outcomes.push(TopLevelCallOutcome {
205                            status: TopLevelCallStatus::Failure,
206                            return_data: state.return_data.clone(),
207                            state,
208                        });
209                        break;
210                    }
211                    StepOutcome::AssumeRejected | StepOutcome::Forked => break,
212                }
213            }
214        }
215
216        Ok(outcomes)
217    }
218
219    pub(super) fn materialize_sequence(
220        &mut self,
221        steps: &[SequenceStepTemplate],
222        state: &PathState,
223    ) -> Result<(Vec<SymbolicInvariantStep>, Vec<SymbolicStorageAssignment>), SymbolicError> {
224        let model = self.solver.model(&mut self.cx, &state.constraints)?;
225        let sequence = steps
226            .iter()
227            .map(|step| {
228                let args = step.calldata.model_to_args(&mut self.cx, &model)?;
229                let calldata = Bytes::from(step.function.abi_encode_input(&args)?);
230                Ok(SymbolicInvariantStep {
231                    sender: step.sender,
232                    address: step.address,
233                    contract_name: step.contract_name.clone(),
234                    function_name: step.function.name.clone(),
235                    signature: step.function.signature(),
236                    args,
237                    calldata,
238                })
239            })
240            .collect::<Result<Vec<_>, SymbolicError>>()?;
241        let storage = state.world.replay_storage_assignments(&model)?;
242        Ok((sequence, storage))
243    }
244}
foundry_evm_symbolic/executor/invariant.rs

foundry_evm_symbolic/executor/
invariant.rs
