Traverse the AST tree from root Visited nodes are placed in a FIFO queue as context to be processed by hook and functions Context defines replacement functions allowing tree to be modified If the tree was modified during the traversal, another pass/traverse is made u
(self, _id=id)
| 172 | self.tree = new_node |
| 173 | |
| 174 | def traverse(self, _id=id): |
| 175 | """ |
| 176 | Traverse the AST tree from root |
| 177 | Visited nodes are placed in a FIFO queue as context to be processed by hook and functions |
| 178 | Context defines replacement functions allowing tree to be modified |
| 179 | If the tree was modified during the traversal, another pass/traverse is made up to specified number of iterations |
| 180 | In case the tree wasn't modified, extra N passes are made as defined by convergence attribute |
| 181 | """ |
| 182 | start = time.time() |
| 183 | self.iteration = 0 |
| 184 | |
| 185 | while self.iteration == 0 or self.modified or self.convergence: |
| 186 | self.queue.clear() |
| 187 | if self.convergence is not None: |
| 188 | # Convergence attribute defines how many extra passes through the tree are made |
| 189 | # after it was not modified, this is a safety mechanism as some badly |
| 190 | # written plugins might not have set modified attribute after modifying the tree |
| 191 | # Or you know, might be a bug and the tree was not marked as modified when it should |
| 192 | if (not self.modified) and self.convergence > 0: |
| 193 | self.convergence -= 1 |
| 194 | else: |
| 195 | # Reset convergence if the tree was modified |
| 196 | self.convergence = 1 |
| 197 | |
| 198 | self.modified = False |
| 199 | |
| 200 | root = self.tree |
| 201 | if type(root) == dict and "ast_tree" in root: |
| 202 | root = root["ast_tree"] |
| 203 | |
| 204 | new_ctx = Context( |
| 205 | node=root, parent=None, replace=self._replace_root, visitor=self |
| 206 | ) |
| 207 | self.queue.append(new_ctx) |
| 208 | self._init_visit(new_ctx) |
| 209 | processed_nodes = set() |
| 210 | |
| 211 | while self.queue: |
| 212 | ctx: Context = self.queue.popleft() |
| 213 | |
| 214 | # Keep track of processed object ID's |
| 215 | # This is to prevent infinite loops where processed object will add themselves back to queue |
| 216 | # This works on python internal ID as we are only concerned about the same objects |
| 217 | if _id(ctx.node) in processed_nodes: |
| 218 | continue |
| 219 | |
| 220 | self.__process_context(ctx) |
| 221 | processed_nodes.add(_id(ctx.node)) |
| 222 | |
| 223 | self._post_iteration() |
| 224 | self.iteration += 1 |
| 225 | if self.iteration >= self.max_iterations: |
| 226 | self.hits.append(Detection( |
| 227 | detection_type="ASTAnalysisError", |
| 228 | message="Maximum AST tree iterations reached", |
| 229 | extra={"iterations": self.iteration}, |
| 230 | signature=f"ast_analysis_error#max_iterations#{str(self.location)}" |
| 231 | )) # TODO: add tests for this |
no test coverage detected