diff --git a/.claude/SCHEMA_DECISIONS.md b/.claude/SCHEMA_DECISIONS.md index f222e9f..12dc408 100644 --- a/.claude/SCHEMA_DECISIONS.md +++ b/.claude/SCHEMA_DECISIONS.md @@ -152,6 +152,17 @@ supported 3.9–3.13+ range; `compute_SSA`'s return contract is discovered from source (not a documented public API) — wrap it behind `ScalpelAliasOracle` to contain upstream drift. +**Follow-up (2026-07-22): Scalpel vendored, now the default.** The Stage-0 +"dependency hygiene" concern proved fatal for a hard dependency: `python-scalpel` +drags `typed_ast`, which has no wheel for Python 3.12+ and does not build from +source there, so `pip install python-scalpel` fails on 3.12/3.13/3.14. Since +`typed_ast` is imported only by `scalpel/typeinfer` (unused here), the 9-module +`SSA`/`cfg`/`core` slice the oracle loads was **vendored** into +`codeanalyzer/dataflow/scalpel/` (Apache-2.0, verbatim but for a `graphviz`-lazy +patch). `ScalpelAliasOracle` is now the shipping default on all supported Python; +`TypeBasedAliasOracle` is the runtime safety net only. See +`docs/superpowers/specs/2026-07-22-vendored-scalpel-default-oracle-design.md`. + ## Stage 5 — keystone conformance sweep (issue #98, schema_version 2.0.0) The stage-5 pre-release conformance check against the canonical schema-v2 diff --git a/CHANGELOG.md b/CHANGELOG.md index 82479b0..e64c020 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,16 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Changed +- The Scalpel-backed L4 points-to oracle is now **vendored** (`typed_ast`-free) + and the default on all supported Python (3.9–3.14); `python-scalpel` is no + longer an optional dependency and the `[scalpel]` extra is removed. On Python + 3.12+ (and anywhere the `[scalpel]` extra was not installed), L4 + `prov:["points-to"]` data-dependence edges are now Scalpel-precise rather than + the coarser type-based over-approximation; the `prov:["ssa"]` set and the + `L3 ⊆ L4` monotonicity invariant are unchanged. Adds `astor` as a runtime + dependency. + ## [1.0.3] - 2026-07-21 ### Fixed diff --git a/CLAUDE.md b/CLAUDE.md index 37a9a05..6816472 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -122,10 +122,13 @@ declared callables by their `can://` tree id, imported/builtin targets by a (`codeanalyzer/dataflow/scalpel_oracle.py`) consumes `python-scalpel`'s **solved SSA + copy/const** state (it never forks the solver) as a copy-closure union-find over access paths, behind the frozen `may_alias(path_a, path_b) -> bool` - interface. `python-scalpel` is an **optional** dependency: if it is absent or a - build/query fails, the analyzer falls back to the total `TypeBasedAliasOracle` - (`alias.py`) and degrades, never raising. The type-based oracle is the sanctioned - fallback, not the shipping default. + interface. Scalpel is **vendored** (`codeanalyzer/dataflow/scalpel/`, a + `typed_ast`-free 9-module slice of `python-scalpel 1.0b0`, Apache-2.0) so it is + the **shipping default** L4 oracle on every supported Python — there is no + external `python-scalpel`/`typed_ast` dependency. `TypeBasedAliasOracle` + (`alias.py`) is retained only as the runtime safety net: on a per-callable + Scalpel build failure or a per-query unresolved access path, `may_alias` + degrades to it, never raising. - **CLI gating** (`codeanalyzer/__main__.py`): `-a` max is 4; `--graphs sdg` requires `-a 4` (a flag error below that); `cfg,dfg,pdg` require `-a 3`; `--graph-field-depth` (the `k_limit`) is valid at L3+. diff --git a/NOTICE b/NOTICE index bd0af50..4ec640c 100644 --- a/NOTICE +++ b/NOTICE @@ -16,3 +16,10 @@ Any other use of CodeQL, including its use on proprietary code or in closed-sour You can view the full CodeQL license at: https://github.com/github/codeql/blob/main/LICENSE + +--- Scalpel License Notice --- + +This project vendors a slice of Scalpel (python-scalpel), a product of SMAT-Lab, +under codeanalyzer/dataflow/scalpel/. Scalpel is licensed under the Apache +License 2.0. The full license text is included at +codeanalyzer/dataflow/scalpel/LICENSE. Source: https://github.com/SMAT-Lab/Scalpel diff --git a/README.md b/README.md index 0a4e8e1..78fccd5 100644 --- a/README.md +++ b/README.md @@ -104,12 +104,9 @@ For the optional **live Neo4j push** (`--emit neo4j --neo4j-uri …`), install t pip install 'codeanalyzer-python[neo4j]' ``` -For the **Scalpel-backed points-to oracle** at level 4, install the `scalpel` extra. It is optional: -when it is absent, level 4 automatically falls back to the built-in type-based oracle. - -```sh -pip install 'codeanalyzer-python[scalpel]' -``` +The **Scalpel-backed points-to oracle** at level 4 is vendored and built in — no extra install +required. If Scalpel cannot resolve a construct, level 4 automatically falls back to the built-in +type-based oracle. ### Install via shell script @@ -529,11 +526,12 @@ symbol-table signature by construction - **Points-to oracle (level 4):** the **Scalpel** may-alias oracle — `ScalpelAliasOracle` (`codeanalyzer/dataflow/scalpel_oracle.py`) — consumes Scalpel's SSA copy/const facts to answer `may_alias(path_a, path_b)`, adding the alias-aware DDG edges (`prov: ["points-to"]`) and the - interprocedural summaries. `python-scalpel` is an **optional dependency** - (`pip install 'codeanalyzer-python[scalpel]'`); when it is absent or cannot resolve a construct, - the analyzer automatically falls back to the built-in `TypeBasedAliasOracle` (Jedi-inferred types; - unknown types conservatively alias), keeping the `may_alias` interface total. Call dispatch comes - from the merged Jedi(+PyCG) call graph, treated as a frozen oracle. + interprocedural summaries. Scalpel is **vendored** — a `typed_ast`-free slice built into the + package under `codeanalyzer/dataflow/scalpel/` — so it is the **default** level-4 oracle with no + external dependency to install; the analyzer falls back to the built-in `TypeBasedAliasOracle` + (Jedi-inferred types; unknown types conservatively alias) only when Scalpel can't resolve a + construct or a per-callable build fails, keeping the `may_alias` interface total. Call dispatch + comes from the merged Jedi(+PyCG) call graph, treated as a frozen oracle. - **Summaries:** relational formal-in → formal-out flows composed bottom-up over the Tarjan SCC condensation of the call graph, a monotone fixpoint within SCCs; globals ride as extra formals, closure captures bind at definition sites. diff --git a/codeanalyzer/dataflow/scalpel/LICENSE b/codeanalyzer/dataflow/scalpel/LICENSE new file mode 100644 index 0000000..261eeb9 --- /dev/null +++ b/codeanalyzer/dataflow/scalpel/LICENSE @@ -0,0 +1,201 @@ + Apache License + Version 2.0, January 2004 + http://www.apache.org/licenses/ + + TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION + + 1. Definitions. + + "License" shall mean the terms and conditions for use, reproduction, + and distribution as defined by Sections 1 through 9 of this document. + + "Licensor" shall mean the copyright owner or entity authorized by + the copyright owner that is granting the License. + + "Legal Entity" shall mean the union of the acting entity and all + other entities that control, are controlled by, or are under common + control with that entity. For the purposes of this definition, + "control" means (i) the power, direct or indirect, to cause the + direction or management of such entity, whether by contract or + otherwise, or (ii) ownership of fifty percent (50%) or more of the + outstanding shares, or (iii) beneficial ownership of such entity. + + "You" (or "Your") shall mean an individual or Legal Entity + exercising permissions granted by this License. + + "Source" form shall mean the preferred form for making modifications, + including but not limited to software source code, documentation + source, and configuration files. + + "Object" form shall mean any form resulting from mechanical + transformation or translation of a Source form, including but + not limited to compiled object code, generated documentation, + and conversions to other media types. + + "Work" shall mean the work of authorship, whether in Source or + Object form, made available under the License, as indicated by a + copyright notice that is included in or attached to the work + (an example is provided in the Appendix below). + + "Derivative Works" shall mean any work, whether in Source or Object + form, that is based on (or derived from) the Work and for which the + editorial revisions, annotations, elaborations, or other modifications + represent, as a whole, an original work of authorship. For the purposes + of this License, Derivative Works shall not include works that remain + separable from, or merely link (or bind by name) to the interfaces of, + the Work and Derivative Works thereof. + + "Contribution" shall mean any work of authorship, including + the original version of the Work and any modifications or additions + to that Work or Derivative Works thereof, that is intentionally + submitted to Licensor for inclusion in the Work by the copyright owner + or by an individual or Legal Entity authorized to submit on behalf of + the copyright owner. For the purposes of this definition, "submitted" + means any form of electronic, verbal, or written communication sent + to the Licensor or its representatives, including but not limited to + communication on electronic mailing lists, source code control systems, + and issue tracking systems that are managed by, or on behalf of, the + Licensor for the purpose of discussing and improving the Work, but + excluding communication that is conspicuously marked or otherwise + designated in writing by the copyright owner as "Not a Contribution." + + "Contributor" shall mean Licensor and any individual or Legal Entity + on behalf of whom a Contribution has been received by Licensor and + subsequently incorporated within the Work. + + 2. Grant of Copyright License. Subject to the terms and conditions of + this License, each Contributor hereby grants to You a perpetual, + worldwide, non-exclusive, no-charge, royalty-free, irrevocable + copyright license to reproduce, prepare Derivative Works of, + publicly display, publicly perform, sublicense, and distribute the + Work and such Derivative Works in Source or Object form. + + 3. Grant of Patent License. Subject to the terms and conditions of + this License, each Contributor hereby grants to You a perpetual, + worldwide, non-exclusive, no-charge, royalty-free, irrevocable + (except as stated in this section) patent license to make, have made, + use, offer to sell, sell, import, and otherwise transfer the Work, + where such license applies only to those patent claims licensable + by such Contributor that are necessarily infringed by their + Contribution(s) alone or by combination of their Contribution(s) + with the Work to which such Contribution(s) was submitted. If You + institute patent litigation against any entity (including a + cross-claim or counterclaim in a lawsuit) alleging that the Work + or a Contribution incorporated within the Work constitutes direct + or contributory patent infringement, then any patent licenses + granted to You under this License for that Work shall terminate + as of the date such litigation is filed. + + 4. Redistribution. You may reproduce and distribute copies of the + Work or Derivative Works thereof in any medium, with or without + modifications, and in Source or Object form, provided that You + meet the following conditions: + + (a) You must give any other recipients of the Work or + Derivative Works a copy of this License; and + + (b) You must cause any modified files to carry prominent notices + stating that You changed the files; and + + (c) You must retain, in the Source form of any Derivative Works + that You distribute, all copyright, patent, trademark, and + attribution notices from the Source form of the Work, + excluding those notices that do not pertain to any part of + the Derivative Works; and + + (d) If the Work includes a "NOTICE" text file as part of its + distribution, then any Derivative Works that You distribute must + include a readable copy of the attribution notices contained + within such NOTICE file, excluding those notices that do not + pertain to any part of the Derivative Works, in at least one + of the following places: within a NOTICE text file distributed + as part of the Derivative Works; within the Source form or + documentation, if provided along with the Derivative Works; or, + within a display generated by the Derivative Works, if and + wherever such third-party notices normally appear. The contents + of the NOTICE file are for informational purposes only and + do not modify the License. You may add Your own attribution + notices within Derivative Works that You distribute, alongside + or as an addendum to the NOTICE text from the Work, provided + that such additional attribution notices cannot be construed + as modifying the License. + + You may add Your own copyright statement to Your modifications and + may provide additional or different license terms and conditions + for use, reproduction, or distribution of Your modifications, or + for any such Derivative Works as a whole, provided Your use, + reproduction, and distribution of the Work otherwise complies with + the conditions stated in this License. + + 5. Submission of Contributions. Unless You explicitly state otherwise, + any Contribution intentionally submitted for inclusion in the Work + by You to the Licensor shall be under the terms and conditions of + this License, without any additional terms or conditions. + Notwithstanding the above, nothing herein shall supersede or modify + the terms of any separate license agreement you may have executed + with Licensor regarding such Contributions. + + 6. Trademarks. This License does not grant permission to use the trade + names, trademarks, service marks, or product names of the Licensor, + except as required for reasonable and customary use in describing the + origin of the Work and reproducing the content of the NOTICE file. + + 7. Disclaimer of Warranty. Unless required by applicable law or + agreed to in writing, Licensor provides the Work (and each + Contributor provides its Contributions) on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or + implied, including, without limitation, any warranties or conditions + of TITLE, NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A + PARTICULAR PURPOSE. You are solely responsible for determining the + appropriateness of using or redistributing the Work and assume any + risks associated with Your exercise of permissions under this License. + + 8. Limitation of Liability. In no event and under no legal theory, + whether in tort (including negligence), contract, or otherwise, + unless required by applicable law (such as deliberate and grossly + negligent acts) or agreed to in writing, shall any Contributor be + liable to You for damages, including any direct, indirect, special, + incidental, or consequential damages of any character arising as a + result of this License or out of the use or inability to use the + Work (including but not limited to damages for loss of goodwill, + work stoppage, computer failure or malfunction, or any and all + other commercial damages or losses), even if such Contributor + has been advised of the possibility of such damages. + + 9. Accepting Warranty or Additional Liability. While redistributing + the Work or Derivative Works thereof, You may choose to offer, + and charge a fee for, acceptance of support, warranty, indemnity, + or other liability obligations and/or rights consistent with this + License. However, in accepting such obligations, You may act only + on Your own behalf and on Your sole responsibility, not on behalf + of any other Contributor, and only if You agree to indemnify, + defend, and hold each Contributor harmless for any liability + incurred by, or claims asserted against, such Contributor by reason + of your accepting any such warranty or additional liability. + + END OF TERMS AND CONDITIONS + + APPENDIX: How to apply the Apache License to your work. + + To apply the Apache License to your work, attach the following + boilerplate notice, with the fields enclosed by brackets "[]" + replaced with your own identifying information. (Don't include + the brackets!) The text should be enclosed in the appropriate + comment syntax for the file format. We also recommend that a + file or class name and description of purpose be included on the + same "printed page" as the copyright notice for easier + identification within third-party archives. + + Copyright [yyyy] [name of copyright owner] + + Licensed under the Apache License, Version 2.0 (the "License"); + you may not use this file except in compliance with the License. + You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + + Unless required by applicable law or agreed to in writing, software + distributed under the License is distributed on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + See the License for the specific language governing permissions and + limitations under the License. diff --git a/codeanalyzer/dataflow/scalpel/README.md b/codeanalyzer/dataflow/scalpel/README.md new file mode 100644 index 0000000..d2c630f --- /dev/null +++ b/codeanalyzer/dataflow/scalpel/README.md @@ -0,0 +1,34 @@ +# Vendored Scalpel (typed_ast-free slice) + +Vendored from **[SMAT-Lab/Scalpel](https://github.com/SMAT-Lab/Scalpel)**, +package `python-scalpel==1.0b0`, licensed **Apache-2.0** (see `LICENSE`). + +## Why vendored + +`python-scalpel` hard-depends on `typed_ast`, whose last release (1.5.5) has no +wheel for Python 3.12+ and fails to build from source on modern compilers — so +`pip install python-scalpel` fails on 3.12/3.13/3.14. `typed_ast` is imported by +exactly one scalpel module, `typeinfer/analysers.py`, which codeanalyzer does not +use. Vendoring the small slice the L4 may-alias oracle needs removes the +`typed_ast` dependency and makes scalpel the default oracle on every supported +Python. + +## What is vendored + +Exactly the 9-module closure that `scalpel.SSA.const` + `scalpel.cfg` load +(verified via `sys.modules`; provably free of `typeinfer`/`typed_ast`): + + __init__.py + SSA/__init__.py, SSA/const.py + cfg/__init__.py, cfg/builder.py, cfg/model.py + core/__init__.py, core/func_call_visitor.py, core/vars_visitor.py + +Copied verbatim except **one patch**: `cfg/model.py`'s top-level +`import graphviz as gv` is guarded (`try/except ImportError: gv = None`) so the +module imports without the `graphviz` package — the graphviz-using +`build_visual()` methods are unused here. + +Runtime deps of this slice: `astor`, `networkx` (both core dependencies of +codeanalyzer). `typed_ast` and `graphviz` are NOT required. + +To refresh: re-run the vendoring in `docs/superpowers/plans/2026-07-22-vendored-scalpel-default-oracle.md` Task 1. diff --git a/codeanalyzer/dataflow/scalpel/SSA/__init__.py b/codeanalyzer/dataflow/scalpel/SSA/__init__.py new file mode 100644 index 0000000..7cf7a86 --- /dev/null +++ b/codeanalyzer/dataflow/scalpel/SSA/__init__.py @@ -0,0 +1,8 @@ +""" +Static Single Assignment (SSA) is a technique of IR in the compiling thoery, it also shows great benefits to static anaysis tasks such as constant propagation, dead code elimination and etc. +Constant propagation is also a matured technique in static anaysis. +It is the process of evaluating or recognizing the actual constant values or expressions at a particular program point. This is realized by utilizing control flow and data flow information. Determining the possible values for variables before runtime gives great benefits to software anaysis. +For instance, with constant value propagation, we can detect and remove dead code or perfrom type checking. +In scalpel, we implement constant propagation along with the SSA for execution efficiency. +""" +__slots__ = ["const"] \ No newline at end of file diff --git a/codeanalyzer/dataflow/scalpel/SSA/const.py b/codeanalyzer/dataflow/scalpel/SSA/const.py new file mode 100644 index 0000000..2b97895 --- /dev/null +++ b/codeanalyzer/dataflow/scalpel/SSA/const.py @@ -0,0 +1,398 @@ +""" +In this module, the single static assignment forms are implemented to allow +further analysis. The module contain a single class named SSA. +""" +import ast +import astor +from functools import reduce +from collections import OrderedDict +import networkx as nx +from ..core.vars_visitor import get_vars + +def parse_val(node): + # does not return anything + if isinstance(node, ast.Constant): + return node.value + if isinstance(node, ast.Str): + if hasattr(node, "value"): + return node.value + else: + return node.s + return "other" + + +class SSA: + """ + Build SSA graph from a given AST node based on the CFG. + """ + def __init__ (self): + """ + Args: + src: the source code as input. + """ + # the class SSA takes a module as the input + self.numbering = {} # numbering variables + self.var_values = {} # numbering variables + self.global_live_idents = [] + self.ssa_blocks = [] + self.error_paths = {} + self.dom = {} + + self.block_ident_gen = {} + self.block_ident_use = {} + self.reachable_table = {} + id2block = {} + self.unreachable_names = {} + self.undefined_names_from = {} + self.global_names = [] + + def get_attribute_stmts(self, stmts): + call_stmts = [] + for stmt in stmts: + if isinstance(stmt,ast.Call) and isinstance(stmt.func, ast.Attribute): + call_stmts += [stmt] + + def get_identifiers(self, ast_node): + """ + Extract all identifiers from the given AST node. + Args: + ast_node: AST node. + """ + if ast_node is None: + return [] + res = get_vars(ast_node) + idents = [r['name'] for r in res if r['name'] is not None and "." not in r['name']] + return idents + + def compute_SSA(self, cfg): + """ + Compute single static assignment form representations for a given CFG. + During the computing, constant value and alias pairs are generated. The following steps are used to compute SSA representations: + step 1a: compute the dominance frontier + step 1b: use dominance frontier to place phi node + if node X contains assignment to a, put phi node for an in dominance frontier of X + adding phi function may require introducing additional phi function + start from the entry node + step2: rename variables so only one definition per name + + Args: + cfg: a control flow graph. + """ + # to count how many times a var is defined + ident_name_counter = {} + # constant assignment dict + ident_const_dict = {} + # step 1a: compute the dominance frontier + all_blocks = cfg.get_all_blocks() + id2blocks = {block.id:block for block in all_blocks} + + block_loaded_idents = {block.id:[] for block in all_blocks} + block_stored_idents = {block.id:[] for block in all_blocks} + + block_const_dict = {block.id:[] for block in all_blocks} + + block_renamed_stored = {block.id:[] for block in all_blocks} + block_renamed_loaded = {block.id:[] for block in all_blocks} + + DF = self.compute_DF(all_blocks) + + for block in all_blocks: + df_nodes = DF[block.id] + tmp_const_dict = {} + + + for idx, stmt in enumerate(block.statements): + stmt_const_dict = {} + stored_idents, loaded_idents, func_names = self.get_stmt_idents_ctx(stmt, const_dict=stmt_const_dict) + tmp_const_dict[idx] = stmt_const_dict + block_loaded_idents[block.id] += [loaded_idents] + block_stored_idents[block.id] += [stored_idents] + block_renamed_loaded[block.id] += [{ident:set() for ident in loaded_idents}] + + block_const_dict[block.id] = tmp_const_dict + + for block in all_blocks: + stored_idents = block_stored_idents[block.id] + loaded_idents = block_loaded_idents[block.id] + n_stmts = len(stored_idents) + assert (n_stmts == len(loaded_idents)) + affected_idents = [] + tmp_const_dict = block_const_dict[block.id] + for i in range(n_stmts): + stmt_stored_idents = stored_idents[i] + stmt_loaded_idents = loaded_idents[i] + stmt_renamed_stored = {} + + for ident in stmt_stored_idents: + affected_idents.append(ident) + if ident in ident_name_counter: + ident_name_counter[ident] += 1 + else: + ident_name_counter[ident] = 0 + # rename the var name as the number of assignments + stmt_const_dict = tmp_const_dict[i] + if ident in stmt_const_dict: + ident_const_dict[(ident, ident_name_counter[ident])] = stmt_const_dict[ident] + + stmt_renamed_stored[ident] = ident_name_counter[ident] + block_renamed_stored[block.id] += [stmt_renamed_stored] + + + #same block, number used identifiers + for ident in stmt_loaded_idents: + # a list of dictions for each of idents used in this statement + phi_loaded_idents = block_renamed_loaded[block.id][i] + if ident in ident_name_counter: + phi_loaded_idents[ident].add(ident_name_counter[ident]) + + df_block_ids = DF[block.id] + for df_block_id in df_block_ids: + df_block = id2blocks[df_block_id] + block_ident_gen_produced = [] + df_block_stored_idents = block_stored_idents[df_block_id] + for af_ident in affected_idents: + # this for-loop process every statement in the block + for idx, phi_loaded_idents in enumerate(block_renamed_loaded[df_block_id]): + block_ident_gen_produced.extend(df_block_stored_idents[idx]) + if af_ident in block_ident_gen_produced: + continue + # place phi function here this var used + # if af_ident has been assigned in this block beforclee this statement, then discard it + # so theck af_ident has been generated in this block + if af_ident in phi_loaded_idents: + phi_loaded_idents[af_ident].add(ident_name_counter[af_ident]) + + return block_renamed_loaded, ident_const_dict + + def get_stmt_idents_ctx(self, stmt, del_set=[], const_dict = {}): + """ + Extract the contextual information of each of identifiers. + For assignment statements, the assigned values for each of variables will be stored. + In addition, the del_set will store all deleted variables. + Args: + stmt: statement from AST trees. + del_set: deleted identifiers + const_dict: a mapping relationship between variables and their assigned values in this statement + """ + # if this is a definition of class/function, ignore + stored_idents = [] + loaded_idents = [] + func_names = [] + # assignment with only one target + + if isinstance(stmt, ast.Assign): + targets = stmt.targets + value = stmt.value + if len(targets) == 1: + if hasattr(targets[0], "id"): + left_name = stmt.targets[0].id + const_dict[left_name] = stmt.value + elif isinstance(targets[0], ast.Attribute): + left_name = astor.to_source(stmt.targets[0]).strip() + const_dict[left_name] = value + # multiple targets are represented as tuple + elif isinstance(targets[0], ast.Tuple): + # value is also represented as tuple + if isinstance(value, ast.Tuple): + for elt, val in zip(targets[0].elts, value.elts): + if hasattr(elt, "id"): + left_name = elt.id + const_dict[left_name] = val + elif isinstance(targets[0], ast.Attribute): + #TODO: resolve attributes + pass + # value is represented as call + if isinstance(value, ast.Call): + for elt in targets[0].elts: + if hasattr(elt, "id"): + left_name = elt.id + const_dict[left_name] = value + elif isinstance(targets[0], ast.Attribute): + #TODO: resolve attributes + pass + else: + # Note in some python versions, there are more than one target for an assignment + # while in some other python versions, multiple targets are deemed as ast.Tuple type in assignment statement + for target in stmt.targets: + # this is an assignment to tuple such as a,b = fun() + # then no valid constant value can be recorded for this statement + if hasattr(target, "id"): + left_name = target.id + const_dict[left_name] = None # TODO: design a type for these kind of values + elif isinstance(stmt.targets[0], ast.Attribute): + #TODO: resolve attributes + pass + + + + # one target assignment with type annotations + if isinstance(stmt, ast.AnnAssign): + if hasattr(stmt.target, "id"): + left_name = stmt.target.id + const_dict[left_name] = stmt.value + elif isinstance(stmt.target, ast.Attribute): + #TODO: resolve attributes + pass + if isinstance(stmt, ast.AugAssign): + # note here , we need to rewrite this value to its extended form + # if the statement is "a += 1", then the assigned value should be a+1 + if hasattr(stmt.target, "id"): + left_name = stmt.target.id + extended_right = ast.BinOp(ast.Name(left_name, ast.Load()), stmt.op, stmt.value) + const_dict[left_name] = extended_right + elif isinstance(stmt.target, ast.Attribute): + #TODO: resolve attributes + pass + if isinstance(stmt, ast.For): + # there is a variation of assignment in for loop + # in the case of : for i in [1,2,3] + # the element of stmt.iter is the value of this assignment + if hasattr(stmt.target, "id"): + left_name = stmt.target.id + iter_value = stmt.iter + # make a iter call + #iter_node = ast.Call(ast.Name("iter", ast.Load()), [stmt.iter], []) + # make a next call + #next_call_node = ast.Call(ast.Name("next", ast.Load()), [iter_node], []) + const_dict[left_name] = iter_value + + elif isinstance(stmt.target, ast.Tuple): + # to handle for-loop uch as: + # for x, y in fun(): + for elt in stmt.target.elts: + if hasattr(elt, "id"): + const_dict[elt.id] = stmt.iter + elif isinstance(stmt.target, ast.Attribute): + #TODO: resolve attributes + pass + + + + if isinstance(stmt, (ast.FunctionDef, ast.AsyncFunctionDef)): + stored_idents.append(stmt.name) + const_dict[stmt.name] = stmt + func_names.append(stmt.name) + new_stmt = stmt + new_stmt.body = [] + ident_info = get_vars(new_stmt) + for r in ident_info: + if r['name'] is None: + continue + if r['usage'] == "load": + loaded_idents.append(r['name']) + return stored_idents, loaded_idents, func_names + + if isinstance(stmt, ast.ClassDef): + stored_idents.append(stmt.name) + const_dict[stmt.name] = None + func_names.append(stmt.name) + return stored_idents, loaded_idents, func_names + + # if this is control flow statements, we should not visit its body to avoid duplicates + # as they are already in the next blocks + if isinstance(stmt, (ast.Import, ast.ImportFrom)): + for alias in stmt.names: + if alias.asname is None: + stored_idents += [alias.name.split('.')[0]] + else: + stored_idents += [alias.asname.split('.')[0]] + return stored_idents, loaded_idents, [] + + if isinstance(stmt, (ast.Try)): + for handler in stmt.handlers: + if handler.name is not None: + stored_idents.append(handler.name) + + if isinstance(handler.type, ast.Name): + loaded_idents.append(handler.type.id) + elif isinstance(handler.type, ast.Attribute) and isinstance(handler.type.value, ast.Name): + loaded_idents.append(handler.type.value.id) + return stored_idents, loaded_idents, [] + if isinstance(stmt, ast.Global): + for name in stmt.names: + self.global_names.append(name) + return stored_idents, loaded_idents, [] + + visit_node = stmt + + if isinstance(visit_node,(ast.If, ast.IfExp)): + # visit_node.body = [] + # visit_node.orlse=[] + visit_node = stmt.test + + elif isinstance(visit_node, (ast.With)): + visit_node.body = [] + visit_node.orlse=[] + + elif isinstance(visit_node, (ast.While)): + visit_node.body = [] + + elif isinstance(visit_node, (ast.For)): + visit_node.body = [] + + elif isinstance(visit_node, ast.Return): + # imaginary variable + stored_idents.append("") + const_dict[""] = visit_node.value + elif isinstance(visit_node, ast.Yield): + # imaginary variable + stored_idents.append("") + const_dict[""] = visit_node.value + + ident_info = get_vars(visit_node) + for r in ident_info: + if r['name'] is None or "_hidden_" in r['name']: + continue + if r['usage'] == 'store': + stored_idents.append(r['name']) + else: + loaded_idents.append(r['name']) + if r['usage'] == 'del': + del_set.append(r['name']) + return stored_idents, loaded_idents, [] + + def to_json(self): + pass + + def print_block(self, block): + return block.get_source() + + # compute the dominators + def compute_idom(self, ssa_blocks): + """ + Compute immediate dominators for each of blocks + Args: + ssa_blocks: blocks from a control flow graph. + """ + # construct the Graph + entry_block = ssa_blocks[0] + G = nx.DiGraph() + for block in ssa_blocks: + G.add_node(block.id) + exits = block.exits + preds = block.predecessors + for link in preds+exits: + G.add_edge(link.source.id, link.target.id) + # DF = nx.dominance_frontiers(G, entry_block.id) + idom = nx.immediate_dominators(G, entry_block.id) + return idom + + # compute dominance frontiers + def compute_DF(self, ssa_blocks): + """ + Compute dominating frontiers for each of blocks + Args: + ssa_blocks: blocks from a control flow graph. + """ + # construct the Graph + entry_block = ssa_blocks[0] + G = nx.DiGraph() + for block in ssa_blocks: + G.add_node(block.id) + exits = block.exits + preds = block.predecessors + for link in preds+exits: + G.add_edge(link.source.id, link.target.id) + DF = nx.dominance_frontiers(G, entry_block.id) + #idom = nx.immediate_dominators(G, entry_block.id) + return DF diff --git a/codeanalyzer/dataflow/scalpel/__init__.py b/codeanalyzer/dataflow/scalpel/__init__.py new file mode 100644 index 0000000..961ac40 --- /dev/null +++ b/codeanalyzer/dataflow/scalpel/__init__.py @@ -0,0 +1,11 @@ +""" +Static Anaysis for Python Programs +================================== +Scalpel is a Python library integrating classical program anaysis algorithms +with tailored features for Python language. It aims to provide simple and +efficient solutions to software engineering researchers that are accessible to +everybody and reusable in various contexts. +""" + +__all__ = ["cfg", "call_graph", "SSA", "core", "typeinfer", "import_graph", "rewriter"] +__version__ = '1.0dev' diff --git a/codeanalyzer/dataflow/scalpel/cfg/__init__.py b/codeanalyzer/dataflow/scalpel/cfg/__init__.py new file mode 100644 index 0000000..ddb2fd5 --- /dev/null +++ b/codeanalyzer/dataflow/scalpel/cfg/__init__.py @@ -0,0 +1,10 @@ +""" +The control-flow graph(CFG) is an essential component in static flow analysis with applications such as program +optimization and taint analysis. +scalpel.cfg module is used to construct the control flow graph for given python programs. The basic unit in the CFG, +Block, contains a list of sequential statements that can be executed in a program without any control jumps. The Blocks +are linked by Link objects, which represent control flow jumps between two blocks and contain the jump conditions in +the form of an expression. Please see the example diagram a control flow graph ![Fibonacci CFG](https://raw.githubusercontent.com/SMAT-Lab/Scalpel/main/docs/_static/resources/cfg_example.png) +""" +from .builder import CFGBuilder +from .model import Block, Link, CFG diff --git a/codeanalyzer/dataflow/scalpel/cfg/builder.py b/codeanalyzer/dataflow/scalpel/cfg/builder.py new file mode 100644 index 0000000..05cb2e5 --- /dev/null +++ b/codeanalyzer/dataflow/scalpel/cfg/builder.py @@ -0,0 +1,700 @@ +""" +This implementation is partly adapted from the static cfg project +https://github.com/coetaur0/staticfg +""" + +import ast +from .model import Block, Link, CFG +import sys +from ..core.func_call_visitor import get_func_calls + + +def is_py38_or_higher(): + if sys.version_info.major == 3 and sys.version_info.minor >= 8: + return True + return False + + +NAMECONSTANT_TYPE = ast.Constant if is_py38_or_higher() else ast.NameConstant + + +def invert(node): + """ + Invert the operation in an ast node object (get its negation). + Args: + node: An ast node object. + Returns: + An ast node object containing the inverse (negation) of the input node. + """ + inverse = {ast.Eq: ast.NotEq, + ast.NotEq: ast.Eq, + ast.Lt: ast.GtE, + ast.LtE: ast.Gt, + ast.Gt: ast.LtE, + ast.GtE: ast.Lt, + ast.Is: ast.IsNot, + ast.IsNot: ast.Is, + ast.In: ast.NotIn, + ast.NotIn: ast.In} + + if type(node) == ast.Compare: + op = type(node.ops[0]) + inverse_node = ast.Compare(left=node.left, ops=[inverse[op]()], + comparators=node.comparators) + elif isinstance(node, ast.BinOp) and type(node.op) in inverse: + op = type(node.op) + inverse_node = ast.BinOp(node.left, inverse[op](), node.right) + elif type(node) == NAMECONSTANT_TYPE and node.value in [True, False]: + inverse_node = NAMECONSTANT_TYPE(value=not node.value) + else: + inverse_node = ast.UnaryOp(op=ast.Not(), operand=node) + return inverse_node + + +def merge_exitcases(exit1, exit2): + """ + Merge the exitcases of two Links. + + Args: + exit1: The exitcase of a Link object. + exit2: Another exitcase to merge with exit1. + + Returns: + The merged exitcases. + """ + if exit1: + if exit2: + return ast.BoolOp(ast.And(), values=[exit1, exit2]) + return exit1 + return exit2 + + +class CFGBuilder(ast.NodeVisitor): + """ + Control flow graph builder. + + A control flow graph builder is an ast.NodeVisitor that can walk through + a program's AST and iteratively build the corresponding CFG. + """ + + __all__ = ["build", "build_from_src", "build_from_file"] + + def __init__(self, separate=False): + super().__init__() + self.after_loop_block_stack = [] + self.curr_loop_guard_stack = [] + self.current_block = None + self.separate_node_blocks = separate + self.enter_func_def = False + + # ---------- CFG building methods ---------- # + def build(self, name, tree, asynchr=False, entry_id=0, flattened=False): + """ + Build a CFG from an AST. + + Args: + name: The name of the CFG being built. + tree: The root of the AST from which the CFG must be built. + async: Boolean indicating whether the CFG being built represents an + asynchronous function or not. When the CFG of a Python + program is being built, it is considered like a synchronous + 'main' function. + entry_id: Value for the id of the entry block of the CFG. + flattened: if use k-v format for all CFGs while hiding its nested information. Key will be fully-qualified names. + + Returns: + The CFG produced from the AST. + """ + self.cfg = CFG(name, asynchr=asynchr) + # Tracking of the current block while building the CFG. + self.current_id = entry_id + self.current_block = self.new_block() + self.cfg.entryblock = self.current_block + # Actual building of the CFG is done here. + self.visit(tree) + visited = [] + self.clean_cfg(self.cfg.entryblock,visited) + + if flattened: + self.cfg = self._flatten_cfg(self.cfg) + pass + return self.cfg + + def _flatten_cfg(self, mod_cfg): + flattend_cfg = {} + + def process_cfg(cfg, dotted_name=["mod"], name_type="mod"): + fully_qualified_name = ".".join(dotted_name) + flattend_cfg[fully_qualified_name] = cfg + for fun_name_tup, fun_cfg in cfg.functioncfgs.items(): + process_cfg(fun_cfg, dotted_name = dotted_name +[fun_name_tup[1]], name_type= "func") + + for cls_name, cls_cfg in cfg.class_cfgs.items(): + process_cfg(cls_cfg, dotted_name = dotted_name +[cls_name], name_type = "cls") + + process_cfg(mod_cfg) + + return flattend_cfg + def build_from_src(self, name, src, flattened=False): + """ + Build a CFG from some Python source code. + + Args: + name: The name of the CFG being built. + src: A string containing the source code to build the CFG from. + flattened: if use k-v format for all CFGs while hiding its nested information. Key will be fully-qualified names. + + + Returns: + The CFG produced from the source code. + """ + tree = ast.parse(src, mode='exec') + return self.build(name, tree, flattened=flattened) + + def build_from_file(self, name, filepath, flattened=False): + """ + Build a CFG from some Python source file. + + Args: + name: The name of the CFG being built. + filepath: The path to the file containing the Python source code to build the CFG from. + flattened: if use k-v format for all CFGs while hiding its nested information. Key will be fully-qualified names. + + + Returns: + The CFG produced from the source file. + """ + with open(filepath, 'r',encoding="utf8") as src_file: + src = src_file.read() + return self.build_from_src(name, src, flattened=flattened) + + # ---------- Graph management methods ---------- # + def new_block(self): + """ + Create a new block with a new id. + Returns: + A Block object with a new unique id. + """ + self.current_id += 1 + return Block(self.current_id) + + def add_statement(self, block, statement): + """ + Add a statement to a block. + Args: + block: A Block object to which a statement must be added. + statement: An AST node representing the statement that must be + added to the current block. + """ + # remove function def nodes + block.statements.append(statement) + + def add_exit(self, block, nextblock, exitcase=None): + """ + Add a new exit to a block. + Args: + block: A block to which an exit must be added. + nextblock: The block to which control jumps from the new exit. + exitcase: An AST node representing the 'case' (or condition) + leading to the exit from the block in the program. + """ + newlink = Link(block, nextblock, exitcase) + block.exits.append(newlink) + nextblock.predecessors.append(newlink) + + def new_loopguard(self): + """ + Create a new block for a loop's guard if the current block is not + empty. Links the current block to the new loop guard. + + Returns: + The block to be used as new loop guard. + """ + if (self.current_block.is_empty() and + len(self.current_block.exits) == 0): + # If the current block is empty and has no exits, it is used as + # entry block (condition test) for the loop. + loopguard = self.current_block + else: + # Jump to a new block for the loop's guard if the current block + # isn't empty or has exits. + loopguard = self.new_block() + self.add_exit(self.current_block, loopguard) + return loopguard + + def new_functionCFG(self, node, asynchr=False, enclosing_block_id=-1): + """ + Create a new sub-CFG for a function definition and add it to the + function CFGs of the CFG being built. + + Args: + node: The AST node containing the function definition. + async: Boolean indicating whether the function for which the CFG is + being built is asynchronous or not. + """ + self.current_id += 1 + # A new sub-CFG is created for the body of the function definition and + # added to the function CFGs of the current CFG. + func_body = ast.Module(body=node.body) + func_builder = CFGBuilder() + self.cfg.functioncfgs[(enclosing_block_id,node.name)] = func_builder.build(node.name, + func_body, + asynchr, + self.current_id) + + def get_arg_names(argument_node): + arg_names = [] + for node in ast.walk(argument_node): + if isinstance(node, ast.arg): + arg_names.append( node.arg) + return arg_names + + self.cfg.function_args[(enclosing_block_id, node.name)] = get_arg_names(node.args) + self.current_id = func_builder.current_id + 1 + + def new_ClassCFG(self, node, asynchr=False): + """ + Create a new sub-CFG for a class definition and add it to the + function CFGs of the CFG being built. + + Args: + node: The AST node containing the function definition. + asynchr: Boolean indicating whether the function for which the CFG is + being built is asynchronous or not. + """ + self.current_id += 1 + # A new sub-CFG is created for the body of the function definition and + # added to the function CFGs of the current CFG. + func_body = ast.Module(body=node.body) + func_builder = CFGBuilder() + base_names = [] + for base in node.bases: + if not isinstance(base, ast.Name): + continue + base_names.append(base.id) + if node.name in self.cfg.class_cfgs and node.name in base_names: + existing_class_cfg = self.cfg.class_cfgs[node.name] + new_class_cfg = func_builder.build(node.name, + func_body, + asynchr, + self.current_id) + new_class_cfg.entryblock.statements = new_class_cfg.entryblock.statements+existing_class_cfg.entryblock.statements + new_class_cfg.functioncfgs.update(existing_class_cfg.functioncfgs) + new_class_cfg.function_args.update(existing_class_cfg.function_args) + self.cfg.class_cfgs[node.name]=new_class_cfg + else: + self.cfg.class_cfgs[node.name] = func_builder.build(node.name, + func_body, + asynchr, + self.current_id) + + self.current_id = func_builder.current_id + 1 + + def clean_cfg(self, block, visited): + """ + Remove the useless (empty) blocks from a CFG. + + Args: + block: The block from which to start traversing the CFG to clean + it. + visited: A list of blocks that already have been visited by + clean_cfg (recursive function). + """ + # Don't visit blocks twice. + if block.id in visited: + return + visited.append(block.id) + + # Empty blocks are removed from the CFG. + if block.is_empty(): + for pred in block.predecessors: + for exit in block.exits: + self.add_exit(pred.source, exit.target, + merge_exitcases(pred.exitcase, + exit.exitcase)) + # Check if the exit hasn't yet been removed from + # the predecessors of the target block. + if exit in exit.target.predecessors: + exit.target.predecessors.remove(exit) + # Check if the predecessor hasn't yet been removed from + # the exits of the source block. + if pred in pred.source.exits: + pred.source.exits.remove(pred) + + block.predecessors = [] + # as the exits may be modified during the recursive call, it is unsafe to iterate on block.exits + # Created a copy of block.exits before calling clean cfg , and iterate over it instead. + for exit in block.exits[:]: + self.clean_cfg(exit.target, visited) + block.exits = [] + else: + for exit in block.exits[:]: + self.clean_cfg(exit.target, visited) + + def goto_new_block(self, node): + if self.separate_node_blocks: + newblock = self.new_block() + self.add_exit(self.current_block, newblock) + self.current_block = newblock + self.generic_visit(node) + + # start visting all statements in AST tree + def visit_Expr(self, node): + self.add_statement(self.current_block, node) + self.goto_new_block(node) + + def visit_Call(self, node): + def visit_func(node): + if type(node) == ast.Name: + return node.id + elif type(node) == ast.Attribute: + # Recursion on series of calls to attributes. + func_name = visit_func(node.value) + func_name += "." + node.attr + return func_name + elif type(node) == ast.Str: + return node.s + elif type(node) == ast.Subscript: + return node.value.id + + #func = node.func + #func_name = visit_func(func) + func_name = get_func_calls(node)[0] + self.current_block.func_calls.append(func_name) + + def visit_Assign(self, node): + self.add_statement(self.current_block, node) + self.goto_new_block(node) + + def visit_AnnAssign(self, node): + self.add_statement(self.current_block, node) + self.goto_new_block(node) + + def visit_AugAssign(self, node): + self.add_statement(self.current_block, node) + self.goto_new_block(node) + + def visit_Global(self, node): + self.add_statement(self.current_block, node) + self.goto_new_block(node) + + def visit_Nonlocal(self, node): + self.add_statement(self.current_block, node) + self.goto_new_block(node) + + def visit_Pass(self, node): + self.add_statement(self.current_block, node) + self.goto_new_block(node) + + def visit_Delete(self, node): + self.add_statement(self.current_block, node) + self.goto_new_block(node) + + def visit_Raise(self, node): + self.add_statement(self.current_block, node) + self.cfg.finalblocks.append(self.current_block) + self.current_block = self.new_block() + + def visit_Assert(self, node): + self.add_statement(self.current_block, node) + # New block for the case in which the assertion 'fails'. + failblock = self.new_block() + self.add_exit(self.current_block, failblock, invert(node.test)) + # If the assertion fails, the current flow ends, so the fail block is a + # final block of the CFG. + self.cfg.finalblocks.append(failblock) + # If the assertion is True, continue the flow of the program. + successblock = self.new_block() + self.add_exit(self.current_block, successblock, node.test) + self.current_block = successblock + self.goto_new_block(node) + + def visit_Try(self, node): + # Add the try statement at the end of the current block. + self.add_statement(self.current_block, node) + + # Create a new block for the body of try. + try_block = self.new_block() + self.add_exit(self.current_block, try_block, ast.Constant(True)) + n_else_stmts = len(node.orelse) + #else_block = self.new_block() + #self.add_exit(self.current_block, try_block, ast.Constant(True)) + + # Create blocks for handlers + n_handlers = len(node.handlers) + handler_blocks = [] + for i in range(n_handlers): + h_block = self.new_block() + handler_blocks += [h_block] + after_try_block = self.new_block() + #self.add_exit(self.current_block, after_try_block, ast.Constant(False)) + # keep the original block + current_block = self.current_block + # + self.current_block = try_block + + for child in node.body: + self.visit(child) + + if n_else_stmts>0: + else_block = self.new_block() + self.add_exit(self.current_block, else_block) + self.current_block = else_block + # create else block + for child in node.orelse: + self.visit(child) + self.add_exit(self.current_block, after_try_block) + + + for i in range(n_handlers): + self.current_block = current_block + handler = node.handlers[i] + self.add_exit(self.current_block, handler_blocks[i], handler.type) + self.current_block = handler_blocks[i] + self.visit(handler) + # If encountered a break, exit will have already been added + if not self.current_block.exits: + self.add_exit(self.current_block, after_try_block) + #self.add_exit(self.current_block, after_try_block) + + #if not self.current_block.exits: + # self.add_exit(self.current_block, after_try_block) + # Continue building the CFG in the after-if block. + + self.current_block = after_try_block + + #populate the block in the try + #self.current_block = try_block + #for child in node.body: + # self.visit(child) + #if not self.current_block.exits: + # self.add_exit(self.current_block, after_try_block) + #self.current_block = after_try_block + + def visit_If(self, node): + # Add the If statement at the end of the current block. + self.add_statement(self.current_block, node) + + # Create a new block for the body of the if. + if_block = self.new_block() + self.add_exit(self.current_block, if_block, node.test) + + # Create a block for the code after the if-else. + afterif_block = self.new_block() + + # New block for the body of the else if there is an else clause. + if len(node.orelse) != 0: + else_block = self.new_block() + self.add_exit(self.current_block, else_block, invert(node.test)) + self.current_block = else_block + # Visit the children in the body of the else to populate the block. + for child in node.orelse: + self.visit(child) + # If encountered a break, exit will have already been added + if not self.current_block.exits: + self.add_exit(self.current_block, afterif_block) + else: + self.add_exit(self.current_block, afterif_block, invert(node.test)) + + # Visit children to populate the if block. + self.current_block = if_block + for child in node.body: + self.visit(child) + if not self.current_block.exits: + self.add_exit(self.current_block, afterif_block) + + # Continue building the CFG in the after-if block. + self.current_block = afterif_block + + + def visit_While(self, node): + loop_guard = self.new_loopguard() + self.current_block = loop_guard + self.add_statement(self.current_block, node) + self.curr_loop_guard_stack.append(loop_guard) + # New block for the case where the test in the while is True. + while_block = self.new_block() + self.add_exit(self.current_block, while_block, node.test) + + # New block for the case where the test in the while is False. + afterwhile_block = self.new_block() + self.after_loop_block_stack.append(afterwhile_block) + inverted_test = invert(node.test) + # Skip shortcut loop edge if while True: + if not (isinstance(inverted_test, NAMECONSTANT_TYPE) and + inverted_test.value is False): + self.add_exit(self.current_block, afterwhile_block, inverted_test) + # Populate the while block. + self.current_block = while_block + for child in node.body: + self.visit(child) + if not self.current_block.exits: + # Did not encounter a break statement, loop back + self.add_exit(self.current_block, loop_guard) + + # Continue building the CFG in the after-while block. + self.current_block = afterwhile_block + self.after_loop_block_stack.pop() + self.curr_loop_guard_stack.pop() + + def visit_For(self, node): + loop_guard = self.new_loopguard() + self.current_block = loop_guard + self.add_statement(self.current_block, node) + self.curr_loop_guard_stack.append(loop_guard) + # New block for the body of the for-loop. + for_block = self.new_block() + self.add_exit(self.current_block, for_block, node.iter) + + # Block of code after the for loop. + afterfor_block = self.new_block() + self.add_exit(self.current_block, afterfor_block) + self.after_loop_block_stack.append(afterfor_block) + self.current_block = for_block + + # Populate the body of the for loop. + for child in node.body: + self.visit(child) + if not self.current_block.exits: + # Did not encounter a break + self.add_exit(self.current_block, loop_guard) + + # Continue building the CFG in the after-for block. + self.current_block = afterfor_block + # Popping the current after loop stack,taking care of errors in case of nested for loops + self.after_loop_block_stack.pop() + self.curr_loop_guard_stack.pop() + + # Async for loops and async with context managers. + # They have the same fields as For and With, respectively. + # Only valid in the body of an AsyncFunctionDef. + # https://docs.python.org/3/library/ast.html + def visit_AsyncFor(self, node): + loop_guard = self.new_loopguard() + self.current_block = loop_guard + self.add_statement(self.current_block, node) + self.curr_loop_guard_stack.append(loop_guard) + # New block for the body of the for-loop. + for_block = self.new_block() + self.add_exit(self.current_block, for_block, node.iter) + + # Block of code after the for loop. + afterfor_block = self.new_block() + self.add_exit(self.current_block, afterfor_block) + self.after_loop_block_stack.append(afterfor_block) + self.current_block = for_block + + # Populate the body of the for loop. + for child in node.body: + self.visit(child) + if not self.current_block.exits: + # Did not encounter a break + self.add_exit(self.current_block, loop_guard) + + # Continue building the CFG in the after-for block. + self.current_block = afterfor_block + # Popping the current after loop stack,taking care of errors in case of nested for loops + self.after_loop_block_stack.pop() + self.curr_loop_guard_stack.pop() + def visit_Break(self, node): + assert len(self.after_loop_block_stack), "Found break not inside loop" + self.add_exit(self.current_block, self.after_loop_block_stack[-1]) + + def visit_Continue(self, node): + assert len(self.curr_loop_guard_stack), "Found continue outside loop" + self.add_exit(self.current_block, self.curr_loop_guard_stack[-1]) + + def visit_Import(self, node): + self.add_statement(self.current_block, node) + + def visit_ImportFrom(self, node): + self.add_statement(self.current_block, node) + + def visit_FunctionDef(self, node): + self.add_statement(self.current_block, node) + self.new_functionCFG(node, asynchr=False, enclosing_block_id=self.current_block.id) + + def visit_AsyncFunctionDef(self, node): + self.add_statement(self.current_block, node) + self.new_functionCFG(node, asynchr=True,enclosing_block_id=self.current_block.id) + + def visit_ClassDef(self, node): + self.add_statement(self.current_block, node) + self.new_ClassCFG(node, asynchr=True) + return node + + def visit_Await(self, node): + afterawait_block = self.new_block() + self.add_exit(self.current_block, afterawait_block) + self.goto_new_block(node) + self.current_block = afterawait_block + + def visit_Return(self, node): + self.add_statement(self.current_block, node) + self.cfg.finalblocks.append(self.current_block) + # Continue in a new block but without any jump to it -> all code after + # the return statement will not be included in the CFG. + self.current_block = self.new_block() + + def visit_Yield(self, node): + self.cfg.asynchr = True + afteryield_block = self.new_block() + self.add_exit(self.current_block, afteryield_block) + self.current_block = afteryield_block + + def visit_With(self, node): + # add with statement to the current block + self.add_statement(self.current_block, node) + # New block for the body of the with. + with_block = self.new_block() + # link current block to with block + self.add_exit(self.current_block, with_block) + + # Block of code after the with. + afterwith_block = self.new_block() + # no branch here + # link with block and body of with + #print(with_block, afterwith_block) + #self.add_exit(with_block, afterwith_block) + # go to with block and create more + self.current_block = with_block + + # Populate the body of the with loop. + for child in node.body: + self.visit(child) + + if not self.current_block.exits: + self.add_exit(self.current_block, afterwith_block) + # Continue building the CFG in the after-with block. + self.current_block = afterwith_block + + # Async for loops and async with context managers. + # They have the same fields as For and With, respectively. + # Only valid in the body of an AsyncFunctionDef. + # https://docs.python.org/3/library/ast.html + def visit_AsyncWith(self, node): + # add with statement to the current block + self.add_statement(self.current_block, node) + # New block for the body of the with. + with_block = self.new_block() + # link current block to with block + self.add_exit(self.current_block, with_block) + + # Block of code after the with. + afterwith_block = self.new_block() + # no branch here + # link with block and body of with + #print(with_block, afterwith_block) + #self.add_exit(with_block, afterwith_block) + # go to with block and create more + self.current_block = with_block + + # Populate the body of the with loop. + for child in node.body: + self.visit(child) + + if not self.current_block.exits: + self.add_exit(self.current_block, afterwith_block) + # Continue building the CFG in the after-with block. + self.current_block = afterwith_block + diff --git a/codeanalyzer/dataflow/scalpel/cfg/model.py b/codeanalyzer/dataflow/scalpel/cfg/model.py new file mode 100644 index 0000000..6191706 --- /dev/null +++ b/codeanalyzer/dataflow/scalpel/cfg/model.py @@ -0,0 +1,331 @@ +""" +Control flow graph for Python programs. +""" + +import ast +import sys +import re +import token +import tokenize +import astor + +try: # PATCH (codeanalyzer): graphviz is only used by build_visual(), which + import graphviz as gv # codeanalyzer never calls. Keep the module importable +except ImportError: # without the graphviz dependency. + gv = None + +__all__ = ["Block", "Link", "CFG"] + +class Block(object): + """ + Basic block in a control flow graph. + + Contains a list of statements executed in a program without any control + jumps. A block of statements is exited through one of its exits. Exits are + a list of Links that represent control flow jumps. + """ + + __slots__ = ["id", "statements", "func_calls", "predecessors", "exits"] + + def __init__(self, id): + # Id of the block. + self.id = id + # Statements in the block. + self.statements = [] + # Calls to functions inside the block (represents context switches to + # some functions' CFGs). + self.func_calls = [] + # Links to predecessors in a control flow graph. + self.predecessors = [] + # Links to the next blocks in a control flow graph. + self.exits = [] + + def __del__(self): + self.statements.clear() + self.func_calls.clear() + self.predecessors.clear() + self.exits.clear() + + def __str__(self): + if self.statements: + return "block:{}@{}".format(self.id, self.at()) + return "empty block:{}".format(self.id) + + def __repr__(self): + txt = "{} with {} exits".format(str(self), len(self.exits)) + if self.statements: + txt += ", body=[" + txt += ", ".join([ast.dump(node) for node in self.statements]) + txt += "]" + return txt + + def at(self): + """ + Get the line number of the first statement of the block in the program. + """ + if self.statements and self.statements[0].lineno >= 0: + return self.statements[0].lineno + return None + + def is_empty(self): + """ + Check if the block is empty. + Returns: + A boolean indicating if the block is empty (True) or not (False). + """ + return len(self.statements) == 0 + ''' + def strip_comment(self, src): + clean_src = "" + + prev_toktype = token.INDENT + first_line = None + last_lineno = -1 + last_col = 0 + + tokgen = tokenize.generate_tokens(src) + for toktype, ttext, (slineno, scol), (elineno, ecol), ltext in tokgen: + if 0: # Change to if 1 to see the tokens fly by. + print("%10s %-14s %-20r %r" % ( + tokenize.tok_name.get(toktype, toktype), + "%d.%d-%d.%d" % (slineno, scol, elineno, ecol), + ttext, ltext + )) + if slineno > last_lineno: + last_col = 0 + if scol > last_col: + mod.write(" " * (scol - last_col)) + if toktype == token.STRING and prev_toktype == token.INDENT: + # Docstring + mod.write("#--") + elif toktype == tokenize.COMMENT: + # Comment + mod.write("##\n") + else: + mod.write(ttext) + prev_toktype = toktype + last_col = ecol + last_lineno = elineno + ''' + def get_source(self): + """ + Get a string containing the Python source code corresponding to the + statements in the block. + Returns: + A string containing the source code of the statements. + """ + src = "#" + str(self.id)+'\n' + for statement in self.statements: + if type(statement) in [ast.If, ast.For, ast.While, ast.With]: + src += (astor.to_source(statement)).split('\n')[0] + "\n" + elif type(statement) == ast.Try: + src += (astor.to_source(statement)).split('\n')[0] + "\n" + #elif type(statement) == ast.If: + # src += (astor.to_source(statement)).split('\n')[0] + "\n" + elif type(statement) in [ast.FunctionDef,ast.AsyncFunctionDef, + ast.ClassDef]: + src += (astor.to_source(statement)).split('\n')[0] + "...\n" + elif type(statement) == ast.ClassDef: + src += (astor.to_source(statement)).split('\n')[0] + "...\n" + else: + src += astor.to_source(statement) + return src + + def get_calls(self): + """ + Get a string containing the calls to other functions inside the block. + + Returns: + A string containing the names of the functions called inside the + block. + """ + txt = "" + for func_call_entry in self.func_calls: + txt += func_call_entry['name'] + '\n' + return txt + + +class Link(object): + """ + Link between blocks in a control flow graph. + + Represents a control flow jump between two blocks. Contains an exitcase in + the form of an expression, representing the case in which the associated + control jump is made. + """ + + __slots__ = ["source", "target", "exitcase"] + + def __init__(self, source, target, exitcase=None): + assert type(source) == Block, "Source of a link must be a block" + assert type(target) == Block, "Target of a link must be a block" + # Block from which the control flow jump was made. + self.source = source + # Target block of the control flow jump. + self.target = target + # 'Case' leading to a control flow jump through this link. + self.exitcase = exitcase + + def __str__(self): + return "link from {} to {}".format(str(self.source), str(self.target)) + + def __repr__(self): + if self.exitcase is not None: + return "{}, with exitcase {}".format(str(self), + ast.dump(self.exitcase)) + return str(self) + + def get_exitcase(self): + """ + Get a string containing the Python source code corresponding to the + exitcase of the Link. + + Returns: + A string containing the source code. + """ + if self.exitcase: + return astor.to_source(self.exitcase) + return "" + def __del__(self): + self.source = None + # Target block of the control flow jump. + self.target = None + # 'Case' leading to a control flow jump through this link. + self.exitcase = None + + + +class CFG(object): + """ + Control flow graph (CFG). + + A control flow graph is composed of basic blocks and links between them + representing control flow jumps. It has a unique entry block and several + possible 'final' blocks (blocks with no exits representing the end of the + CFG). + """ + def __init__(self, name, asynchr=False): + """ + The constructor of CFG class. Only name of this graph is required. + """ + assert type(name) == str, "Name of a CFG must be a string" + assert type(asynchr) == bool, "Async must be a boolean value" + # Name of the function or module being represented. + self.name = name + # Type of function represented by the CFG (sync or async). A Python + # program is considered as a synchronous function (main). + self.asynchr = asynchr + # Entry block of the CFG. + self.entryblock = None + # Final blocks of the CFG. + self.finalblocks = [] + # Sub-CFGs for functions defined inside the current CFG. + self.functioncfgs = {} + self.class_cfgs = {} + + self.function_args = {} + self.class_args = {} + + def __del__(self): + pass + + def __str__(self): + return "CFG for {}".format(self.name) + + def remove_comments(self, src): + pass + + + + def get_all_blocks(self): + """ + Get a list of code blocks in this CFG; This is generated by BFS order. + + Returns: + A list of code blocks. + """ + import queue + all_blocks = [] + + visited = set() + working_queue = queue.Queue() + working_queue.put(self.entryblock) + + while not working_queue.empty(): + block = working_queue.get() + # this block has been visited + if block.id in visited: + continue + all_blocks.append(block) + visited.add(block.id) + for suc_link in block.exits: + if suc_link.target.id not in visited: + working_queue.put(suc_link.target) + return all_blocks + #def dfs(start_block): + # # non-recurisve implementation of DFS search + + def __iter__(self): + """ + Generator that yields all the blocks in the current graph, then + recursively yields from any sub graphs + """ + visited = set() + to_visit = [self.entryblock] + + while to_visit: + block = to_visit.pop(0) + visited.add(block) + for exit_ in block.exits: + if exit_.target in visited or exit_.target in to_visit: + continue + to_visit.append(exit_.target) + yield block + + for subcfg in self.functioncfgs.values(): + yield from subcfg + + def _visit_blocks(self, graph, block, visited=[], calls=True): + # Don't visit blocks twice. + if block.id in visited: + return + + nodelabel = block.get_source() + graph.node(str(block.id), label=nodelabel) + + visited.append(block.id) + # Show the block's function calls in a node. + if calls and block.func_calls: + calls_node = str(block.id)+"_calls" + calls_label = block.get_calls().strip() + graph.node(calls_node, label=calls_label, + _attributes={'shape': 'box'}) + graph.edge(str(block.id), calls_node, label="calls", + _attributes={'style': 'dashed'}) + # Recursively visit all the blocks of the CFG. + for exit in block.exits: + self._visit_blocks(graph, exit.target, visited, calls=calls) + edgelabel = exit.get_exitcase().strip() + graph.edge(str(block.id), str(exit.target.id), label=edgelabel) + + def _build_visual(self, format='pdf', calls=True): + graph = gv.Digraph(name='cluster'+self.name, format=format, + graph_attr={'label': self.name}) + self._visit_blocks(graph, self.entryblock, visited=[], calls=False) + return graph + + def build_visual(self, format, calls=True, show=True): + """ + Build a visualisation of the CFG with graphviz and output it in a DOT + file. + + Args: + filename: The name of the output file in which the visualisation + must be saved. + format: The format to use for the output file (PDF, ...). + show: A boolean indicating whether to automatically open the output + file after building the visualisation. + """ + graph = self._build_visual(format, calls) + return graph + diff --git a/codeanalyzer/dataflow/scalpel/core/__init__.py b/codeanalyzer/dataflow/scalpel/core/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/codeanalyzer/dataflow/scalpel/core/func_call_visitor.py b/codeanalyzer/dataflow/scalpel/core/func_call_visitor.py new file mode 100644 index 0000000..336fc13 --- /dev/null +++ b/codeanalyzer/dataflow/scalpel/core/func_call_visitor.py @@ -0,0 +1,234 @@ +import sys +import ast +from collections import deque +from ast import NodeVisitor +from copy import deepcopy + + +def is_py38_or_higher(): + if sys.version_info.major == 3 and sys.version_info.minor >= 8: + return True + return False + + +NAMECONSTANT_TYPE = ast.Constant if is_py38_or_higher() else ast.NameConstant + + +class CallTransformer(ast.NodeTransformer): + def __init__(self): + self.call_names = [] + + def visit_Attribute(self, node): + # self.generic_visit(node.value) + return node + + def param2str(self, param): + + def get_func(node): + if type(node) == ast.Name: + return node.id + elif type(node) == ast.Constant: + # ingore such as "this is a constant".join() + return "" + elif type(node) == ast.BinOp: + # ingore such as (a+b+c).fun() + return "" + elif type(node) == ast.Str: + # ingore such as "xxx".fun() + return "" + elif type(node) == ast.JoinedStr: + # ingore such as "xxx".fun() + return "" + elif type(node) == ast.Bytes: + # ingore such as "xxx".fun() + return "" + elif type(node) == ast.Compare: + # example "(x.matrix_exp() == torch.eye(20, 20, dtype=dtype, device=device)).all().item()" + # tests/test-cases/cfg-tests/pytorch-test-test_linalg.py + # ignore for now + return "" + elif type(node) == ast.Subscript: + # currently, we will ignore the slices because we cannot track the type of the value. + # for instance, a[something].fun() -> a.fun() + # this sacrifice + return get_func(node.value) + #elif type(node) == ast.JoinedStr: + # return "" + elif type(node) == ast.Attribute: + if type(node.value) in [ast.JoinedStr, ast.Constant]: + return node.attr + else: + return get_func(node.value) + "." + node.attr + elif type(node) == ast.Call: + return get_func(node.func) + elif type(node) == ast.IfExp: + return "" + elif type(node) == ast.Compare: + return "" + elif type(node) == ast.UnaryOp: + return "" + #ast.UnaryOp + else: + #import astor + #print(astor.to_source(node)) + raise Exception(str(type(node))) + + if isinstance(param, ast.Subscript): + return self.param2str(param.value) + if isinstance(param, ast.Call): + return get_func(param) + elif isinstance(param, ast.Name): + return param.id + elif isinstance(param, ast.Num): + # python 3.6 + return param.n + #return param.value + elif isinstance(param, ast.List): + return "List" + elif isinstance(param, ast.ListComp): + return "List" + elif isinstance(param, ast.Tuple): + return "Tuple" + elif isinstance(param, (ast.Dict, ast.DictComp)): + return "Dict" + elif isinstance(param, (ast.Set, ast.SetComp)): + return "Set" + elif isinstance(param, ast.Str): + return param.s + elif isinstance(param, ast.NameConstant): + return param.value + elif isinstance(param, ast.Constant): + return param.value + elif isinstance(param, ast.Expr): + return "Expr" + else: + return "unknown" + + def visit_Call(self, node): + + tmp_fun_node = deepcopy(node) + tmp_fun_node.args = [] + tmp_fun_node.keywords = [] + + callvisitor = FuncCallVisitor() + callvisitor.visit(tmp_fun_node) + + call_info = {"name": callvisitor.name, + "lineno": tmp_fun_node.lineno, + "col_offset": tmp_fun_node.col_offset, + "params": [] + } + self.call_names += [call_info] + for arg in node.args: + call_info["params"] += [self.param2str(arg)] + self.generic_visit(arg) + + for kw in node.keywords: + call_info["params"] += [self.param2str(kw.value)] + self.generic_visit(kw) + self.generic_visit(tmp_fun_node) + + return node + + +class FuncCallVisitor(ast.NodeVisitor): + def __init__(self): + self._name = deque() + self.call_names = [] + + def clear(self): + self._name = deque() + self.call_names = [] + + @property + def name(self): + return '.'.join(self._name) + + @name.deleter + def name(self): + self._name.clear() + + def visit_Name(self, node): + self._name.appendleft(node.id) + + def visit_Attribute(self, node): + + try: + self._name.appendleft(node.attr) + self._name.appendleft(node.value.id) + except AttributeError as e: + self.generic_visit(node) + + def visit_Call(self, node): + node.args = [] + node.keywords = [] + self.generic_visit(node) + return node + + + def visit_Subscript(self, node): + # ingore subscription slice + self.visit(node.value) + return node + +def get_args(node): + arg_type = [] + for arg in node.args: + if isinstance(arg, ast.Name): + arg_type.append(arg.id) + elif isinstance(arg, ast.Num): + arg_type.append("Num") + elif isinstance(arg, ast.List): + arg_type.append("List") + elif isinstance(arg, ast.ListComp): + arg_type.append("List") + elif isinstance(arg, ast.Tuple): + arg_type.append("Tuple") + elif isinstance(arg, ast.Dict): + arg_type.append("Dict") + elif isinstance(arg, ast.DictComp): + arg_type.append("Dict") + elif isinstance(arg, ast.Set): + arg_type.append("Set") + elif isinstance(arg, ast.SetComp): + arg_type.append("Set") + elif isinstance(arg, ast.Str): + arg_type.append("Str") + elif isinstance(arg, ast.NameConstant): + arg_type.append("NameConstant") + elif isinstance(arg, ast.Constant): + arg_type.append("Constant") + elif isinstance(arg, ast.Call): + arg_type.append(("Call", get_func_calls(arg)[0])) + else: + arg_type.append("Other") + return arg_type + + +def get_call_type(tree): + # how to remove + func_calls = [] + for node in ast.walk(tree): + if isinstance(node, ast.Call): + callvisitor = FuncCallVisitor() + callvisitor.visit(node.func) + func_calls += [(callvisitor.name, get_args(node))] + return func_calls + + +def get_call_type(tree): + # how to remove + func_calls = [] + for node in ast.walk(tree): + if isinstance(node, ast.Call): + callvisitor = FuncCallVisitor() + callvisitor.visit(node.func) + func_calls += [(callvisitor.name, get_args(node))] + return func_calls + + +def get_func_calls(tree): + node = deepcopy(tree) + transformer = CallTransformer() + transformer.visit(node) + return transformer.call_names diff --git a/codeanalyzer/dataflow/scalpel/core/vars_visitor.py b/codeanalyzer/dataflow/scalpel/core/vars_visitor.py new file mode 100644 index 0000000..f5e64f2 --- /dev/null +++ b/codeanalyzer/dataflow/scalpel/core/vars_visitor.py @@ -0,0 +1,205 @@ +import ast + +class VarsVisitor(ast.NodeVisitor): + def __init__(self): + self.result = list() + + def _ctx2str(self, ctx): + if isinstance(ctx, ast.Load): + return "load" + elif isinstance(ctx, ast.Store): + return "store" + elif isinstance(ctx, ast.Del): + return "del" + else: + raise Exception("unknown variable context") + + def visit_Name(self, node): + var_info = { + "name": node.id, + "lineno": node.lineno, + "col_offset": node.col_offset, + "usage" : self._ctx2str(node.ctx) + } + + self.result.append(var_info) + + + def visit_BoolOp(self, node): + for v in node.values: + self.visit(v) + + def visit_BinOp(self, node): + self.visit(node.left) + self.visit(node.right) + + def visit_UnaryOp(self, node): + self.visit(node.operand) + + def visit_Lambda(self, node): + return node + + def visit_IfExp(self, node): + self.visit(node.test) + self.visit(node.body) + self.visit(node.orelse) + + def visit_Dict(self, node): + for k in node.keys: + if k is not None: + self.visit(k) + for v in node.values: + self.visit(v) + + def visit_Set(self, node): + for e in node.elts: + self.visit(e) + + def comprehension(self, node): + self.visit(node.target) + self.visit(node.iter) + for c in node.ifs: + self.visit(c) + + def visit_ListComp(self, node): + for gen in node.generators: + self.comprehension(gen) + self.visit(node.elt) + + def visit_SetComp(self, node): + self.visit(node.elt) + for gen in node.generators: + self.comprehension(gen) + + def visit_DictComp(self, node): + for gen in node.generators: + self.comprehension(gen) + self.visit(node.key) + self.visit(node.value) + + + def visit_GeneratorComp(self, node): + self.visit(node.elt) + for gen in node.generators: + self.comprehension(gen) + + def visit_Yield(self, node): + if node.value: + self.visit(node.value) + + def visit_YieldFrom(self, node): + self.visit(node.value) + + def visit_Compare(self, node): + self.visit(node.left) + for c in node.comparators: + self.visit(c) + + def visit_Call(self, node): + self.visit(node.func) + for arg in node.args: + self.visit(arg) + for keyword in node.keywords: + self.visit(keyword) + + def visit_keyword(self, node): + self.visit(node.value) + + def visit_Attribute(self, node): + full_name = self.get_attr_name(node) + + var_info = { + "name": full_name, + "lineno": node.lineno, + "col_offset": node.col_offset, + "usage" : self._ctx2str(node.ctx) + } + self.result.append(var_info) + self.visit(node.value) + + def get_attr_name (self, node): + if isinstance(node, ast.Call): + # to be test + return self.get_attr_name(node.func) + if isinstance(node, ast.Name): + return node.id + elif isinstance(node, ast.Attribute): + attr_name = self.get_attr_name(node.value) + if attr_name is None: + return None + return attr_name +"."+node.attr + elif isinstance(node, ast.Subscript): + return self.get_attr_name(node.value) + else: + # such as (a**2).sum() + return None + + def slicev(self, node): + + if isinstance(node, ast.Constant): + return + if isinstance(node, ast.Slice): + if node.lower: + self.visit(node.lower) + if node.upper: + self.visit(node.upper) + if node.step: + self.visit(node.step) + elif isinstance(node, ast.ExtSlice): + if node.dims: + for d in node.dims: + self.visit(d) + elif isinstance(node,ast.Tuple): + for elt in node.elts: + self.visit(elt) + elif isinstance(node, ast.UnaryOp): + self.visit(node.operand) + elif isinstance(node, ast.Name): + self.visit(node) + # this is due to syntax change + elif hasattr(node, "value"): + self.visit(node.value) + + def visit_Subscript(self, node): + if isinstance(node.value, ast.Attribute): + pass + self.visit(node.value) + self.slicev(node.slice) + + def visit_Starred(self, node): + self.visit(node.value) + + def visit_List(self, node): + for el in node.elts: + self.visit(el) + + def visit_Tuple(self, node): + for el in node.elts: + self.visit(el) + + def visit_FunctionDef(self, node): + + for stmt in node.body: + self.visit(stmt) + #return node + + def visit_Assign(self, node): + for target in node.targets: + #if isinstance(target, ast.Subscript): + # target.value.ctx = ast.Store() + self.visit(target) + if not isinstance(node.value, ast.Lambda): + self.visit(node.value) + + #for target in node.targets: + # if isinstance(target, ast.Subscript): + # target.value.ctx = ast.Store() + # self.visit(target) + #else: + # self.visit(target) + + +def get_vars(node): + visitor = VarsVisitor() + visitor.visit(node) + return visitor.result diff --git a/codeanalyzer/dataflow/scalpel_oracle.py b/codeanalyzer/dataflow/scalpel_oracle.py index 1e25325..5b11e03 100644 --- a/codeanalyzer/dataflow/scalpel_oracle.py +++ b/codeanalyzer/dataflow/scalpel_oracle.py @@ -22,7 +22,7 @@ forks or re-runs Scalpel's solver — and turns the copy/const records into per-function copy-closure equivalence classes: - ``from scalpel.SSA.const import SSA`` + ``from codeanalyzer.dataflow.scalpel.SSA.const import SSA`` ``ssa_results, const_dict = SSA().compute_SSA(func_cfg)`` ``const_dict`` maps ``(name, version)`` to the ``ast`` value node that defined @@ -138,14 +138,14 @@ def from_function( ) -> "ScalpelAliasOracle": """Build from a function AST by consuming Scalpel's solved SSA state. - Imports Scalpel lazily (``ImportError`` if the optional dependency is - absent) and reuses the *same source* both graphs are built from — the + Imports the vendored Scalpel slice (``codeanalyzer.dataflow.scalpel``) + and reuses the *same source* both graphs are built from — the function's unparsed text — so the join is identity, not a fuzzy match. Raises on any build failure; :func:`make_alias_oracle` is the total, never-raising entry point callers should prefer. """ - from scalpel.SSA.const import SSA - from scalpel.cfg import CFGBuilder + from codeanalyzer.dataflow.scalpel.SSA.const import SSA + from codeanalyzer.dataflow.scalpel.cfg import CFGBuilder src = ast.unparse(func_ast) fname = name or getattr(func_ast, "name", None) @@ -250,19 +250,16 @@ def may_alias(self, path_a: str, path_b: str) -> bool: def make_alias_oracle(pycallable, func_ast, base_types) -> object: """Total selector for the L4 may-alias oracle. - Returns a :class:`ScalpelAliasOracle` when ``python-scalpel`` is importable - *and* builds successfully on ``func_ast``; otherwise logs once (INFO) and - returns a :class:`TypeBasedAliasOracle` over ``base_types``. Never raises — - mirrors how ``core._get_pycg_call_graph`` degrades on a missing/failed PyCG. + Returns a :class:`ScalpelAliasOracle` built on the vendored, typed_ast-free + Scalpel slice (``codeanalyzer.dataflow.scalpel``) — the default L4 oracle. + Falls back to :class:`TypeBasedAliasOracle` only when the per-callable + Scalpel build fails on this AST. Never raises. """ fallback = TypeBasedAliasOracle(base_types) try: return ScalpelAliasOracle.from_function( func_ast, base_types=base_types, fallback=fallback ) - except ImportError: - _note_fallback("python-scalpel not installed") - return fallback except Exception: _note_fallback("scalpel alias build failed") logger.debug("scalpel alias oracle build error", exc_info=True) diff --git a/docs/superpowers/plans/2026-07-22-vendored-scalpel-default-oracle.md b/docs/superpowers/plans/2026-07-22-vendored-scalpel-default-oracle.md new file mode 100644 index 0000000..18520bb --- /dev/null +++ b/docs/superpowers/plans/2026-07-22-vendored-scalpel-default-oracle.md @@ -0,0 +1,455 @@ +# Vendored typed_ast-free Scalpel as Default L4 Oracle — Implementation Plan + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** Vendor a minimal, `typed_ast`-free slice of `python-scalpel 1.0b0` into `codeanalyzer/dataflow/scalpel/` and make `ScalpelAliasOracle` the shipping-default L4 oracle on every supported Python, with no external `python-scalpel`/`typed_ast` dependency. + +**Architecture:** Copy the 9-module `SSA`/`cfg`/`core` closure the oracle actually loads (verified free of `typeinfer`/`typed_ast`) into a `codeanalyzer.dataflow.scalpel` package, patched only to make `graphviz` a non-required import. Repoint `scalpel_oracle.py` at the vendored path and drop its "optional dependency absent" fallback branch, so the type-based oracle demotes to a pure runtime safety net. + +**Tech Stack:** Python 3.9–3.14, `astor` (new runtime dep), `networkx` (existing), pytest, Poetry (dev env, currently 3.14.5), uv (CI + fetching the vendored source). + +## Global Constraints + +- Conventional Commits (`type(scope): summary`). +- NEVER add AI/Claude authorship anywhere (no `Co-Authored-By`, "Generated with", 🤖) — commits, PRs, code, docs. Absolute. +- **Apache-2.0 attribution is mandatory** for the vendored code: copy upstream `LICENSE` into the vendored dir, add a provenance `README.md`, and add a `NOTICE` entry. +- **`requires-python` stays `>=3.9`** — do NOT cap it. +- **Behavior:** the monotonicity invariant `L3 ⊆ L4` must still hold (the `prov:["ssa"]` edge set is unchanged; scalpel only sharpens the additive `prov:["points-to"]` overlay). No schema change (`schema_version` stays `2.0.0`). +- The vendored copy is **verbatim** from `python-scalpel 1.0b0` except the single documented `graphviz` patch. +- Vendor **exactly** these 9 files — no other scalpel modules (`typeinfer`, `call_graph`, `pycg`, `import_graph`, `scope_graph`, `dataflow`, `rewriter.py`, and the unused `SSA/{alg,def_use,ssa}.py` + extra `core/*.py`): + `__init__.py`, `SSA/__init__.py`, `SSA/const.py`, `cfg/__init__.py`, `cfg/builder.py`, `cfg/model.py`, `core/__init__.py`, `core/func_call_visitor.py`, `core/vars_visitor.py`. +- Run tests with `poetry run python -m pytest` (managed env, py3.14). Do NOT use `uv run` for the dev env (it creates a stray in-project `.venv` that shadows Poetry). There must be no in-project `.venv`. + +## File Structure + +- Create `codeanalyzer/dataflow/scalpel/` — the 9 vendored files + `LICENSE` + `README.md`. +- Modify `codeanalyzer/dataflow/scalpel/cfg/model.py` — the one `graphviz`-lazy patch. +- Modify `codeanalyzer/dataflow/scalpel_oracle.py` — repoint imports, drop the dead ImportError branch. +- Modify `pyproject.toml` — add `astor`, remove the `[scalpel]` extra. +- Modify `NOTICE` — Scalpel Apache-2.0 attribution. +- Modify `CLAUDE.md`, `.claude/SCHEMA_DECISIONS.md`, `CHANGELOG.md` — record the reversal + behavior change. +- Create `test/test_vendored_scalpel.py` — import-hygiene, typed_ast-free, fidelity, determinism tests. + +--- + +### Task 1: Vendor the scalpel slice + dependency + attribution + +**Files:** +- Create: `codeanalyzer/dataflow/scalpel/**` (9 files + `LICENSE` + `README.md`) +- Modify: `codeanalyzer/dataflow/scalpel/cfg/model.py` (graphviz patch) +- Modify: `pyproject.toml` (add `astor`, remove `[scalpel]` extra) +- Modify: `NOTICE` +- Test: `test/test_vendored_scalpel.py` + +**Interfaces:** +- Produces: importable `codeanalyzer.dataflow.scalpel.SSA.const.SSA` and `codeanalyzer.dataflow.scalpel.cfg.CFGBuilder`, functional with no external `scalpel`/`typed_ast`/`graphviz`. + +- [ ] **Step 1: Write the failing import-hygiene test** + +Create `test/test_vendored_scalpel.py`: + +```python +"""The vendored, typed_ast-free scalpel slice: it must import and compute SSA +with no external scalpel / typed_ast / graphviz, and never pull in typeinfer.""" +import sys +import importlib + + +def test_vendored_scalpel_imports_and_computes_ssa_typed_ast_free(): + # No external scalpel shadowing the vendored copy. + assert "scalpel" not in sys.modules or not sys.modules["scalpel"].__file__.endswith( + "site-packages/scalpel/__init__.py" + ), "external python-scalpel is installed; the test must exercise the vendored copy" + + from codeanalyzer.dataflow.scalpel.cfg import CFGBuilder + from codeanalyzer.dataflow.scalpel.SSA.const import SSA + + src = "def f(a):\n b = a\n c = b\n return c\n" + module_cfg = CFGBuilder().build_from_src("m", src) + func_cfg = list(module_cfg.functioncfgs.values())[0] + ssa_results, const_dict = SSA().compute_SSA(func_cfg) + # The copy chain + return pseudo-name scalpel is known to produce. + assert ("b", 0) in const_dict and ("c", 0) in const_dict and ("", 0) in const_dict + + # The whole point: no typed_ast, and typeinfer was never vendored/loaded. + assert "typed_ast" not in sys.modules + loaded = [m for m in sys.modules if m.startswith("codeanalyzer.dataflow.scalpel")] + assert not any("typeinfer" in m for m in loaded), f"typeinfer leaked: {loaded}" +``` + +- [ ] **Step 2: Run it to confirm it fails** + +Run: `poetry run python -m pytest test/test_vendored_scalpel.py -q` +Expected: FAIL with `ModuleNotFoundError: No module named 'codeanalyzer.dataflow.scalpel'`. + +- [ ] **Step 3: Vendor the 9 files + LICENSE (reproducible fetch)** + +Fetch the exact upstream source and copy only the 9 files + the license: + +```bash +cd /home/rkrsn/workspace/codellm-devkit/codeanalyzer-python +TMP=$(mktemp -d) +uv pip install --no-deps --target "$TMP" 'python-scalpel==1.0b0' +DST=codeanalyzer/dataflow/scalpel +mkdir -p "$DST/SSA" "$DST/cfg" "$DST/core" +cp "$TMP/scalpel/__init__.py" "$DST/__init__.py" +cp "$TMP/scalpel/SSA/__init__.py" "$DST/SSA/__init__.py" +cp "$TMP/scalpel/SSA/const.py" "$DST/SSA/const.py" +cp "$TMP/scalpel/cfg/__init__.py" "$DST/cfg/__init__.py" +cp "$TMP/scalpel/cfg/builder.py" "$DST/cfg/builder.py" +cp "$TMP/scalpel/cfg/model.py" "$DST/cfg/model.py" +cp "$TMP/scalpel/core/__init__.py" "$DST/core/__init__.py" +cp "$TMP/scalpel/core/func_call_visitor.py" "$DST/core/func_call_visitor.py" +cp "$TMP/scalpel/core/vars_visitor.py" "$DST/core/vars_visitor.py" +cp "$TMP"/python_scalpel-1.0b0.dist-info/LICENSE "$DST/LICENSE" +rm -rf "$TMP" +``` + +Verify exactly 9 `.py` files landed: + +```bash +find codeanalyzer/dataflow/scalpel -name '*.py' | sort +# expect the 9 listed in Global Constraints, nothing else +``` + +- [ ] **Step 4: Apply the single graphviz patch** + +In `codeanalyzer/dataflow/scalpel/cfg/model.py`, the top-level `import graphviz as gv` (line 12) makes the module require `graphviz` at load. `graphviz` is used only by the visualization methods (`_build_visual`/`build_visual`), which codeanalyzer never calls. Replace the bare import: + +```python +import graphviz as gv +``` + +with a guarded import so the module loads without `graphviz`: + +```python +try: # PATCH (codeanalyzer): graphviz is only used by build_visual(), which + import graphviz as gv # codeanalyzer never calls. Keep the module importable +except ImportError: # without the graphviz dependency. + gv = None +``` + +This is the ONLY change to any vendored file. + +- [ ] **Step 5: Add attribution — `README.md` + `NOTICE` entry** + +Create `codeanalyzer/dataflow/scalpel/README.md`: + +```markdown +# Vendored Scalpel (typed_ast-free slice) + +Vendored from **[SMAT-Lab/Scalpel](https://github.com/SMAT-Lab/Scalpel)**, +package `python-scalpel==1.0b0`, licensed **Apache-2.0** (see `LICENSE`). + +## Why vendored + +`python-scalpel` hard-depends on `typed_ast`, whose last release (1.5.5) has no +wheel for Python 3.12+ and fails to build from source on modern compilers — so +`pip install python-scalpel` fails on 3.12/3.13/3.14. `typed_ast` is imported by +exactly one scalpel module, `typeinfer/analysers.py`, which codeanalyzer does not +use. Vendoring the small slice the L4 may-alias oracle needs removes the +`typed_ast` dependency and makes scalpel the default oracle on every supported +Python. + +## What is vendored + +Exactly the 9-module closure that `scalpel.SSA.const` + `scalpel.cfg` load +(verified via `sys.modules`; provably free of `typeinfer`/`typed_ast`): + + __init__.py + SSA/__init__.py, SSA/const.py + cfg/__init__.py, cfg/builder.py, cfg/model.py + core/__init__.py, core/func_call_visitor.py, core/vars_visitor.py + +Copied verbatim except **one patch**: `cfg/model.py`'s top-level +`import graphviz as gv` is guarded (`try/except ImportError: gv = None`) so the +module imports without the `graphviz` package — the graphviz-using +`build_visual()` methods are unused here. + +Runtime deps of this slice: `astor`, `networkx` (both core dependencies of +codeanalyzer). `typed_ast` and `graphviz` are NOT required. + +To refresh: re-run the vendoring in `docs/superpowers/plans/2026-07-22-vendored-scalpel-default-oracle.md` Task 1. +``` + +Append to the top-level `NOTICE` (after the existing content): + +``` +--- Scalpel License Notice --- + +This project vendors a slice of Scalpel (python-scalpel), a product of SMAT-Lab, +under codeanalyzer/dataflow/scalpel/. Scalpel is licensed under the Apache +License 2.0. The full license text is included at +codeanalyzer/dataflow/scalpel/LICENSE. Source: https://github.com/SMAT-Lab/Scalpel +``` + +- [ ] **Step 6: Add `astor` dep, remove the `[scalpel]` extra** + +In `pyproject.toml`, add to the core `dependencies` array (after the `uv` entry), with a comment: + +```toml + # astor: runtime dependency of the vendored Scalpel SSA slice + # (scalpel/SSA/const.py uses astor.to_source). Pure-Python, installs + # everywhere; scalpel pins ~=0.8.1. + "astor>=0.8.1,<0.9.0", +``` + +Remove the entire `[project.optional-dependencies].scalpel` group (the +`scalpel = ["python-scalpel>=1.0b0"]` block and its comment). Leave the `neo4j` +extra intact. + +- [ ] **Step 7: Install the new dep into the dev env** + +Run: `poetry install --with test` +Then: `poetry run python -c "import astor; print('astor', astor.__version__)"` +Expected: prints the astor version (installs on 3.14 — pure Python). + +- [ ] **Step 8: Run the hygiene test — now green** + +Run: `poetry run python -m pytest test/test_vendored_scalpel.py -q` +Expected: PASS (the vendored slice imports + computes SSA on 3.14, typed_ast-free). + +- [ ] **Step 9: Commit** + +```bash +git add codeanalyzer/dataflow/scalpel pyproject.toml NOTICE test/test_vendored_scalpel.py +git commit -m "feat(dataflow): vendor typed_ast-free scalpel SSA/cfg slice" +``` + +--- + +### Task 2: Repoint the oracle at the vendored slice and make it the default + +**Files:** +- Modify: `codeanalyzer/dataflow/scalpel_oracle.py` +- Test: `test/test_vendored_scalpel.py` (append) + +**Interfaces:** +- Consumes: the vendored `codeanalyzer.dataflow.scalpel.SSA.const.SSA` / `...cfg.CFGBuilder` (Task 1). +- Produces: `make_alias_oracle(pycallable, func_ast, base_types)` returns a `ScalpelAliasOracle` by default (never the type-based oracle for *absence*, only for a per-callable build failure). + +- [ ] **Step 1: Write the failing "scalpel is the default" test** + +Append to `test/test_vendored_scalpel.py`: + +```python +import ast + + +def test_make_alias_oracle_defaults_to_scalpel_without_typed_ast(): + import sys + assert "typed_ast" not in sys.modules + from codeanalyzer.dataflow.scalpel_oracle import make_alias_oracle, ScalpelAliasOracle + + src = "def f(a):\n b = a\n c = b\n return c\n" + func_ast = ast.parse(src).body[0] + oracle = make_alias_oracle(pycallable=None, func_ast=func_ast, base_types={}) + # The whole point: scalpel is the default, not the type-based fallback. + assert isinstance(oracle, ScalpelAliasOracle), type(oracle).__name__ + # It answers queries (copies alias; unrelated locals do not). + assert oracle.may_alias("b", "c") is True + + +def test_make_alias_oracle_is_deterministic(): + import ast as _ast + from codeanalyzer.dataflow.scalpel_oracle import make_alias_oracle + src = "def f(a):\n b = a\n c = b\n return c\n" + fa = _ast.parse(src).body[0] + o1 = make_alias_oracle(None, _ast.parse(src).body[0], {}) + o2 = make_alias_oracle(None, _ast.parse(src).body[0], {}) + pairs = [("a", "b"), ("b", "c"), ("a", "c"), ("b", "b")] + assert [o1.may_alias(x, y) for x, y in pairs] == [o2.may_alias(x, y) for x, y in pairs] +``` + +- [ ] **Step 2: Run it to confirm it fails** + +Run: `poetry run python -m pytest test/test_vendored_scalpel.py -k make_alias_oracle -q` +Expected: FAIL — `make_alias_oracle` still imports `from scalpel.SSA.const import SSA` (external, absent on 3.14) and returns the type-based fallback, so `isinstance(..., ScalpelAliasOracle)` is False. + +- [ ] **Step 3: Repoint the imports in `from_function`** + +In `codeanalyzer/dataflow/scalpel_oracle.py`, in `ScalpelAliasOracle.from_function`, change: + +```python + from scalpel.SSA.const import SSA + from scalpel.cfg import CFGBuilder +``` + +to: + +```python + from codeanalyzer.dataflow.scalpel.SSA.const import SSA + from codeanalyzer.dataflow.scalpel.cfg import CFGBuilder +``` + +Update that method's docstring line "Imports Scalpel lazily (``ImportError`` if the optional dependency is absent)…" to "Imports the vendored Scalpel slice (``codeanalyzer.dataflow.scalpel``) and reuses the *same source*…". Also update the module docstring reference at the top of the file (`from scalpel.SSA.const import SSA`) to the vendored path. + +- [ ] **Step 4: Drop the dead ImportError branch in `make_alias_oracle`** + +Replace the current `make_alias_oracle` body: + +```python +def make_alias_oracle(pycallable, func_ast, base_types) -> object: + """Total selector for the L4 may-alias oracle. + + Returns a :class:`ScalpelAliasOracle` built on the vendored, typed_ast-free + Scalpel slice (``codeanalyzer.dataflow.scalpel``) — the default L4 oracle. + Falls back to :class:`TypeBasedAliasOracle` only when the per-callable + Scalpel build fails on this AST. Never raises. + """ + fallback = TypeBasedAliasOracle(base_types) + try: + return ScalpelAliasOracle.from_function( + func_ast, base_types=base_types, fallback=fallback + ) + except Exception: + _note_fallback("scalpel alias build failed") + logger.debug("scalpel alias oracle build error", exc_info=True) + return fallback +``` + +(The `except ImportError: _note_fallback("python-scalpel not installed"); return fallback` branch is removed — the vendored import cannot be absent. Keep the per-query `self._fallback.may_alias(...)` inside `may_alias` unchanged.) + +- [ ] **Step 5: Run the oracle tests — now green** + +Run: `poetry run python -m pytest test/test_vendored_scalpel.py -q` +Expected: PASS (scalpel is now the default oracle; determinism holds). + +- [ ] **Step 6: L4 regression — the live acceptance test (scalpel-L4 on 3.14)** + +Run: `poetry run python -m pytest test/test_v2_l4.py test/test_v2_l4_summary.py test/test_dataflow_sdg.py test/test_dataflow_defuse.py -q` +Expected: PASS. These now exercise the vendored **scalpel** path (previously the type-based fallback on 3.14). If a case fails, investigate: a genuine scalpel-vs-type-based points-to difference is expected and the test's expectation may need updating to the (correct, tighter) scalpel result — but confirm it's a precision tightening, not a regression, before changing any assertion. If unsure, report rather than edit the assertion. + +- [ ] **Step 7: Commit** + +```bash +git add codeanalyzer/dataflow/scalpel_oracle.py test/test_vendored_scalpel.py +git commit -m "feat(dataflow): make vendored scalpel the default L4 alias oracle" +``` + +--- + +### Task 3: Fidelity check + documentation + +**Files:** +- Test: `test/test_vendored_scalpel.py` (append the fidelity check) +- Modify: `CLAUDE.md`, `.claude/SCHEMA_DECISIONS.md`, `CHANGELOG.md` + +**Interfaces:** +- Consumes: the vendored slice (Task 1) and the rewired oracle (Task 2). Produces no new code interface. + +- [ ] **Step 1: Add the vendored-vs-upstream fidelity test (skipif no pip scalpel)** + +Append to `test/test_vendored_scalpel.py`: + +```python +import pytest + + +def _pip_scalpel_available(): + try: + import importlib.util + # the EXTERNAL package, not our vendored copy + return importlib.util.find_spec("scalpel") is not None and \ + not importlib.util.find_spec("scalpel").origin.endswith( + "codeanalyzer/dataflow/scalpel/__init__.py") + except Exception: + return False + + +@pytest.mark.skipif(not _pip_scalpel_available(), + reason="upstream python-scalpel not installed (3.12+ can't build typed_ast)") +def test_vendored_ssa_matches_upstream(): + """On a Python where pip python-scalpel installs (<=3.11), the vendored copy + must produce identical SSA const_dict keys as upstream on the same source.""" + import scalpel.SSA.const as up_ssa + import scalpel.cfg as up_cfg + from codeanalyzer.dataflow.scalpel.SSA.const import SSA as VSSA + from codeanalyzer.dataflow.scalpel.cfg import CFGBuilder as VCFG + + src = "def f(a):\n b = a\n if a:\n b = 2\n return b\n" + up_c = up_cfg.CFGBuilder().build_from_src("m", src) + v_c = VCFG().build_from_src("m", src) + up_fn = list(up_c.functioncfgs.values())[0] + v_fn = list(v_c.functioncfgs.values())[0] + _, up_const = up_ssa.SSA().compute_SSA(up_fn) + _, v_const = VSSA().compute_SSA(v_fn) + assert sorted(map(str, up_const)) == sorted(map(str, v_const)) +``` + +- [ ] **Step 2: Run the fidelity test (skips on 3.14; that's expected)** + +Run: `poetry run python -m pytest test/test_vendored_scalpel.py -q` +Expected: PASS with the fidelity case SKIPPED on the 3.14 dev env (pip scalpel not installable). + +- [ ] **Step 3: Update `CLAUDE.md`** + +In `CLAUDE.md`, find the "**L4 points-to oracle = Scalpel.**" bullet and replace its "optional dependency … sanctioned fallback, not the shipping default" framing. Change the sentence: + +> `python-scalpel` is an **optional** dependency: if it is absent or a build/query fails, the analyzer falls back to the total `TypeBasedAliasOracle` (`alias.py`) and degrades, never raising. The type-based oracle is the sanctioned fallback, not the shipping default. + +to: + +> Scalpel is **vendored** (`codeanalyzer/dataflow/scalpel/`, a `typed_ast`-free 9-module slice of `python-scalpel 1.0b0`, Apache-2.0) so it is the **shipping default** L4 oracle on every supported Python — there is no external `python-scalpel`/`typed_ast` dependency. `TypeBasedAliasOracle` (`alias.py`) is retained only as the runtime safety net: on a per-callable Scalpel build failure or a per-query unresolved access path, `may_alias` degrades to it, never raising. + +- [ ] **Step 4: Update `.claude/SCHEMA_DECISIONS.md`** + +Append a follow-up note to the "## Stage 0 — Scalpel oracle spike" section: + +```markdown +**Follow-up (2026-07-22): Scalpel vendored, now the default.** The Stage-0 +"dependency hygiene" concern proved fatal for a hard dependency: `python-scalpel` +drags `typed_ast`, which has no wheel for Python 3.12+ and does not build from +source there, so `pip install python-scalpel` fails on 3.12/3.13/3.14. Since +`typed_ast` is imported only by `scalpel/typeinfer` (unused here), the 9-module +`SSA`/`cfg`/`core` slice the oracle loads was **vendored** into +`codeanalyzer/dataflow/scalpel/` (Apache-2.0, verbatim but for a `graphviz`-lazy +patch). `ScalpelAliasOracle` is now the shipping default on all supported Python; +`TypeBasedAliasOracle` is the runtime safety net only. See +`docs/superpowers/specs/2026-07-22-vendored-scalpel-default-oracle-design.md`. +``` + +- [ ] **Step 5: Update `CHANGELOG.md`** + +Under the `## [Unreleased]` heading, add: + +```markdown +### Changed +- The Scalpel-backed L4 points-to oracle is now **vendored** (`typed_ast`-free) + and the default on all supported Python (3.9–3.14); `python-scalpel` is no + longer an optional dependency and the `[scalpel]` extra is removed. On Python + 3.12+ (and anywhere the `[scalpel]` extra was not installed), L4 + `prov:["points-to"]` data-dependence edges are now Scalpel-precise rather than + the coarser type-based over-approximation; the `prov:["ssa"]` set and the + `L3 ⊆ L4` monotonicity invariant are unchanged. Adds `astor` as a runtime + dependency. +``` + +- [ ] **Step 6: Run the full behavior-preservation gate** + +Run: `poetry run python -m pytest test/test_vendored_scalpel.py test/test_v2_l4.py test/test_v2_l4_summary.py test/test_dataflow_sdg.py test/test_dataflow_defuse.py test/test_v2_superset.py -q` +Expected: PASS (incl. the `test_l3_subset_of_l4` monotonicity gate in `test_v2_superset.py`). Note: `test_v2_superset.py` runs the CLI with `--no-venv`; it exercises the vendored scalpel on the L4 fixture. + +- [ ] **Step 7: Commit** + +```bash +git add test/test_vendored_scalpel.py CLAUDE.md .claude/SCHEMA_DECISIONS.md CHANGELOG.md +git commit -m "docs(dataflow): record vendored scalpel as the default L4 oracle" +``` + +--- + +## Self-Review + +**1. Spec coverage:** +- Vendored `codeanalyzer/dataflow/scalpel/` 9-file slice, verbatim + graphviz patch → Task 1 Steps 3–4. ✓ +- Attribution (LICENSE, README, NOTICE) → Task 1 Steps 3, 5. ✓ +- `astor` added, `[scalpel]` extra removed, `requires-python` unchanged → Task 1 Step 6. ✓ +- Oracle repointed + default (ImportError branch dropped, fallback kept) → Task 2 Steps 3–4. ✓ +- Behavior change (L4 points-to precision), monotonicity preserved → Task 2 Step 6, Task 3 Steps 5–6. ✓ +- Docs: CLAUDE.md, SCHEMA_DECISIONS, CHANGELOG → Task 3 Steps 3–5. ✓ +- Testing: import hygiene, typed_ast-free gate, default-scalpel, determinism, fidelity, L4 regression, monotonicity → Tasks 1–3. ✓ + +**2. Placeholder scan:** No TBD/TODO. The vendoring uses exact `cp` commands and an explicit file list; the graphviz patch shows the exact before/after; every doc edit gives the exact old→new text. + +**3. Type consistency:** `make_alias_oracle(pycallable, func_ast, base_types)` signature matches the existing call site in `core`/the pipeline. Import paths (`codeanalyzer.dataflow.scalpel.SSA.const`, `...cfg`) are consistent across Task 1 (creation), Task 2 (oracle import), and the tests. `ScalpelAliasOracle`/`TypeBasedAliasOracle` names match `scalpel_oracle.py`/`alias.py`. diff --git a/docs/superpowers/specs/2026-07-22-vendored-scalpel-default-oracle-design.md b/docs/superpowers/specs/2026-07-22-vendored-scalpel-default-oracle-design.md new file mode 100644 index 0000000..fb0ac43 --- /dev/null +++ b/docs/superpowers/specs/2026-07-22-vendored-scalpel-default-oracle-design.md @@ -0,0 +1,198 @@ +# Vendored, typed_ast-free Scalpel as the default L4 oracle + +- **Date:** 2026-07-22 +- **Status:** Design approved; ready for an implementation plan +- **Scope:** Make the Scalpel-backed points-to oracle (`ScalpelAliasOracle`) the + shipping default for L4 dataflow on **every** supported Python, by vendoring a + minimal, `typed_ast`-free slice of `python-scalpel` into the package. + +## Motivation + +`python-scalpel` is the primary L4 may-alias oracle, but it ships today as an +**optional** extra with a type-based fallback — because `python-scalpel 1.0b0` +hard-depends on **`typed_ast`**, an abandoned package whose last release +(`1.5.5`) has no wheel for Python 3.12+ and fails to compile from source on a +modern compiler. So on the interpreters most users (and this repo's own dev env) +now run — 3.12, 3.13, 3.14 — scalpel cannot be installed at all, and L4 silently +degrades to the coarser type-based oracle. + +We want scalpel to be the **default** L4 oracle everywhere. It cannot be a hard +PyPI dependency (the `typed_ast` wall), and a Python-version cap to reach it +(≤3.11) would drop the 3.12–3.14 support the analyzer just added. The way +through is to **vendor** the small slice of scalpel we use, minus the one module +that drags `typed_ast`. + +## Feasibility (verified, not assumed) + +- **`typed_ast` is spurious for our use.** It is imported in exactly one scalpel + module — `scalpel/typeinfer/analysers.py` — which codeanalyzer never touches. + The oracle uses only `scalpel.SSA.const` and `scalpel.cfg`. +- **The slice runs without `typed_ast`.** On Python 3.12 with `typed_ast` *not* + installed, `cfg.CFGBuilder` + `SSA.const.SSA` import and `compute_SSA` returns + the exact SSA/const facts the oracle consumes (verified: a copy chain yields + `('b',0), ('c',0), ('',0)`). +- **Exact closure = 9 modules**, provably free of `typeinfer`/`typed_ast` + (checked via `sys.modules` after importing the two entry points): + `scalpel/__init__.py`, `SSA/{__init__,const}.py`, + `cfg/{__init__,builder,model}.py`, `core/{__init__,func_call_visitor,vars_visitor}.py`. +- **Third-party imports touched:** `astor` (used at runtime — `SSA/const.py:191` + `astor.to_source`), `graphviz` (imported at module load in `cfg/model.py` but + only used by the unused `build_visual()`), `networkx` (already a dependency). +- **License = Apache-2.0** (confirmed on `SMAT-Lab/Scalpel` and in the wheel's + `LICENSE`) — vendorable with attribution, same family as the already-bundled + PyCG. +- **Upstream is abandoned** (`1.0b0`, frozen) — the usual "vendored copy drifts + from upstream" cost does not apply. + +## Goals + +- `ScalpelAliasOracle` is the **default** L4 oracle on Python 3.9–3.14+, with no + external `python-scalpel`/`typed_ast` dependency and no `pip install` breakage. +- The type-based oracle (`TypeBasedAliasOracle`) is retained purely as the + runtime safety net (per-callable build failure, per-query unresolved path). +- Behavior for users who already had the `[scalpel]` extra on ≤3.11 is unchanged + (same code). Behavior on 3.12+ *improves* (real points-to instead of the + type-based over-approximation). + +## Non-goals + +- No change to the L4 *schema* (`prov` values `ssa`/`points-to` unchanged; the + DDG shape is the same). +- No `requires-python` change — stays `>=3.9`. +- No vendoring of scalpel's `typeinfer`, `call_graph`, `pycg`, `import_graph`, + `scope_graph`, or `dataflow` packages — only the 9-module `SSA`/`cfg`/`core` + closure the oracle actually loads. +- No change to `defuse.py`'s two-rule DDG contract or the `k_limit` machinery. + +## Design + +### The vendored package: `codeanalyzer/dataflow/scalpel/` + +A normal package (no `_vendor/` layer), mirroring upstream's layout so scalpel's +relative imports (`from ..core.vars_visitor import get_vars`) resolve within +`codeanalyzer.dataflow.scalpel`: + +``` +codeanalyzer/dataflow/scalpel/ + __init__.py + SSA/__init__.py, const.py + cfg/__init__.py, builder.py, model.py + core/__init__.py, func_call_visitor.py, vars_visitor.py + LICENSE # upstream Apache-2.0, copied verbatim + README.md # provenance + the one patch (see below) +``` + +The `scalpel/` wrapper name is kept (rather than flattening `SSA`/`cfg`/`core` +into `dataflow/`) because `codeanalyzer/dataflow/cfg.py` already exists and +scalpel ships its own `cfg/` package — the wrapper namespaces them apart. + +The 9 files are copied **verbatim** from `python-scalpel 1.0b0`, with **exactly +one patch**: + +- `cfg/model.py`: the module-load `import graphviz as gv` (line 12) is moved to a + lazy import *inside* the unused `build_visual()` method. This removes `graphviz` + as a runtime dependency (we never render). No other line changes; the + SSA/CFG-computation code paths are byte-identical to upstream. + +### Attribution (Apache-2.0) + +- `codeanalyzer/dataflow/scalpel/LICENSE` — upstream's Apache-2.0 license, copied. +- `codeanalyzer/dataflow/scalpel/README.md` — records: source repo + (`SMAT-Lab/Scalpel`), the pinned version (`1.0b0`), the exact 9-file list, the + single `graphviz`-lazy patch, and that `typeinfer` (the sole `typed_ast` user) + was deliberately excluded. +- The repo's top-level `NOTICE` gains a one-line entry crediting vendored Scalpel + (Apache-2.0), alongside the existing attributions. + +### Dependencies (`pyproject.toml`) + +- **Add** `astor` to core `dependencies` (genuine runtime dep; pure-Python; + installs on every platform/Python). +- **Remove** the `[project.optional-dependencies].scalpel` extra (scalpel is now + built in). +- `networkx` unchanged (already core); `graphviz` and `typed_ast` are **not** + dependencies. +- `requires-python` stays `>=3.9`. + +### Oracle rewiring: `codeanalyzer/dataflow/scalpel_oracle.py` + +- `ScalpelAliasOracle.from_function` imports from the vendored path: + `from codeanalyzer.dataflow.scalpel.SSA.const import SSA` and + `from codeanalyzer.dataflow.scalpel.cfg import CFGBuilder`. +- `make_alias_oracle`: the `ImportError` ("python-scalpel not installed") branch + becomes dead — the vendored import cannot be absent — and is **removed**. The + per-callable **build-failure** `except` (returns the type-based fallback) and + the per-query unresolved-path fallback inside `may_alias` both **stay**, so the + total, never-raises contract is preserved. `ScalpelAliasOracle` is now the + default the selector returns. +- Docstrings/comments updated to drop the "optional / not installed" framing. + +### Behavior change + +On Python 3.12+/3.14 — and for anyone who never installed the old `[scalpel]` +extra — L4's `prov:["points-to"]` DDG edges were previously derived from the +*type-based* over-approximation; they now become **scalpel-precise** (a tighter +subset). The `prov:["ssa"]` edges (the L3 subset) are unchanged, so the +**monotonicity invariant `L3 ⊆ L4` still holds** (the ssa set is untouched; +points-to is an additive overlay). Users who already had the `[scalpel]` extra on +≤3.11 see no change. This L4 precision improvement is recorded in `CHANGELOG.md`. + +### Docs + +`CLAUDE.md` and `.claude/SCHEMA_DECISIONS.md` are updated to record: scalpel is +now **vendored** (`codeanalyzer/dataflow/scalpel/`, `typed_ast`-free) and the +**shipping default** L4 oracle on all supported Python — superseding the earlier +"the type-based oracle is the sanctioned fallback, not the shipping default" +statement (it is now the runtime safety net, not the default). The Stage-0 +`SCHEMA_DECISIONS` entry gains a follow-up noting the `typed_ast` wall on 3.12+ +that forced vendoring. + +## Testing plan + +1. **Import hygiene.** `import codeanalyzer.dataflow.scalpel` and building the + oracle succeed with no external `scalpel`/`typed_ast`/`graphviz` installed; + assert `typeinfer` is never imported (scan `sys.modules` after building the + oracle). +2. **`typed_ast`-free property gate.** Assert `typed_ast` is not importable in + the environment, yet `make_alias_oracle` builds a `ScalpelAliasOracle` and + `may_alias` returns verdicts on a copy-chain fixture — the whole point, + encoded as a regression gate. +3. **Vendored-fidelity check.** On Python ≤3.11 (where pip `python-scalpel` + installs), compare the vendored `compute_SSA`/CFG output against the + pip-installed scalpel on a set of fixtures — proves the copy is faithful. + `skipif` the pip package can't install (3.12+); the copy is verbatim, so this + is a belt-and-suspenders check. +4. **L4 regression (the live acceptance test).** `test_v2_l4.py`, + `test_v2_l4_summary.py`, `test_dataflow_sdg.py`, `test_dataflow_defuse.py` now + exercise the **scalpel** path by default — the first time scalpel-L4 runs on + the 3.14 dev env. They must stay green and now genuinely test scalpel-derived + points-to rather than the fallback. +5. **Determinism.** Run L4 twice on a fixture and assert identical output (the + scalpel SSA feeds `may_alias`; the `#99` `PYTHONHASHSEED=0` pin + the oracle's + access-path normalization must keep it stable). + +## Risks and rollback + +- **Incomplete closure** — a missed transitive import would surface as an + `ImportError` at oracle build, which the per-callable fallback swallows into a + silent type-based degrade. Mitigation: the import-hygiene test (1) builds the + oracle and asserts it is a `ScalpelAliasOracle`, so a broken closure fails + loudly rather than degrading silently. +- **Vendored-copy divergence from upstream behavior** — mitigated by the verbatim + copy (only the `graphviz` import patched) and the fidelity check (3). +- **L4 output change on 3.12+** is intended, documented in the CHANGELOG, and + bounded by the preserved monotonicity invariant. +- **Rollback** is a single revert: delete `codeanalyzer/dataflow/scalpel/`, + restore the `[scalpel]` extra + the `make_alias_oracle` ImportError branch, and + drop `astor`. No schema or id changes to unwind. + +## Out of scope + +- Replacing `astor.to_source` with stdlib `ast.unparse` to drop the `astor` dep + (a behavior-risking change to vendored SSA logic — deferred; `astor` is tiny + and universal). +- A scalpel-installed CI lane / capturing the L4 (`a4`) equivalence goldens from + the separate analysis-pipeline branch — that work belongs to that branch once + it merges; here scalpel simply becomes available for it. +- Vendoring or using scalpel's type-inference (`typeinfer`) for the type-guided + alias branch — the type-based oracle already covers that fallback. diff --git a/pyproject.toml b/pyproject.toml index 69820b4..d36dece 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -53,6 +53,10 @@ dependencies = [ # Shipped as a self-contained binary in its wheel, so it's available wherever # canpy is pip-installed (incl. Docker); core.py falls back to pip without it. "uv>=0.5.0", + # astor: runtime dependency of the vendored Scalpel SSA slice + # (scalpel/SSA/const.py uses astor.to_source). Pure-Python, installs + # everywhere; scalpel pins ~=0.8.1. + "astor>=0.8.1,<0.9.0", ] [project.optional-dependencies] @@ -61,12 +65,6 @@ dependencies = [ neo4j = [ "neo4j>=5.0.0,<6.0.0", ] -# The Scalpel-backed L4 may-alias oracle (ScalpelAliasOracle). Optional: when -# absent, `make_alias_oracle` degrades to the type-based fallback, so analysis -# still runs — installing it only sharpens alias precision. -scalpel = [ - "python-scalpel>=1.0b0", -] [dependency-groups] test = [ diff --git a/test/test_v2_l4.py b/test/test_v2_l4.py index 016f12e..82bcb60 100644 --- a/test/test_v2_l4.py +++ b/test/test_v2_l4.py @@ -5,8 +5,6 @@ import sys from pathlib import Path -import pytest - from codeanalyzer.dataflow.alias import TypeBasedAliasOracle from codeanalyzer.dataflow.identity import IdentityMap from codeanalyzer.dataflow.sdg import ParamNode @@ -104,18 +102,19 @@ def emit(self, record): self.records.append(record) -def test_make_alias_oracle_falls_back_when_scalpel_absent(monkeypatch): +def test_make_alias_oracle_falls_back_on_build_failure(monkeypatch): """`make_alias_oracle` degrades to TypeBasedAliasOracle behavior and logs - once when Scalpel cannot be imported — regardless of whether the optional - dependency is actually installed in the runner.""" + once when the (now-vendored, always-present) Scalpel oracle fails to build + for a given callable — the only surviving fallback path now that Scalpel + can no longer be "absent" (it is vendored, not an optional import).""" import codeanalyzer.dataflow.scalpel_oracle as so - # Force `import scalpel...` to raise ImportError even if it is installed: - # a None entry in sys.modules makes the import machinery halt. Cover the - # top-level package and every submodule the oracle imports so a previously - # cached submodule cannot satisfy the import. - for mod in ("scalpel", "scalpel.cfg", "scalpel.SSA", "scalpel.SSA.const"): - monkeypatch.setitem(sys.modules, mod, None) + # Force the per-callable Scalpel build to fail, regardless of input. + monkeypatch.setattr( + so.ScalpelAliasOracle, + "from_function", + classmethod(lambda cls, *a, **k: (_ for _ in ()).throw(RuntimeError("boom"))), + ) # Reset the process-wide "logged once" guard and capture on the actual # (propagate=False) codeanalyzer logger. @@ -144,9 +143,9 @@ def test_make_alias_oracle_falls_back_when_scalpel_absent(monkeypatch): def test_scalpel_oracle_copy_chain(): - """When Scalpel is importable, the copy chain `b = a; c = b` places a, b, c - in one copy-closure class: a/b may-alias, an unrelated name does not.""" - pytest.importorskip("scalpel") + """Scalpel is vendored and always importable now, so the copy chain + `b = a; c = b` always places a, b, c in one copy-closure class: a/b + may-alias, an unrelated name does not.""" from codeanalyzer.dataflow.scalpel_oracle import ScalpelAliasOracle func_ast = ast.parse(COPY_CHAIN).body[0] diff --git a/test/test_vendored_scalpel.py b/test/test_vendored_scalpel.py new file mode 100644 index 0000000..f789199 --- /dev/null +++ b/test/test_vendored_scalpel.py @@ -0,0 +1,84 @@ +"""The vendored, typed_ast-free scalpel slice: it must import and compute SSA +with no external scalpel / typed_ast / graphviz, and never pull in typeinfer.""" +import ast +import sys + +import pytest + + +def test_vendored_scalpel_imports_and_computes_ssa_typed_ast_free(): + # No external scalpel shadowing the vendored copy. + assert "scalpel" not in sys.modules or not sys.modules["scalpel"].__file__.endswith( + "site-packages/scalpel/__init__.py" + ), "external python-scalpel is installed; the test must exercise the vendored copy" + + from codeanalyzer.dataflow.scalpel.cfg import CFGBuilder + from codeanalyzer.dataflow.scalpel.SSA.const import SSA + + src = "def f(a):\n b = a\n c = b\n return c\n" + module_cfg = CFGBuilder().build_from_src("m", src) + func_cfg = list(module_cfg.functioncfgs.values())[0] + ssa_results, const_dict = SSA().compute_SSA(func_cfg) + # The copy chain + return pseudo-name scalpel is known to produce. + assert ("b", 0) in const_dict and ("c", 0) in const_dict and ("", 0) in const_dict + + # The whole point: no typed_ast, and typeinfer was never vendored/loaded. + assert "typed_ast" not in sys.modules + loaded = [m for m in sys.modules if m.startswith("codeanalyzer.dataflow.scalpel")] + assert not any("typeinfer" in m for m in loaded), f"typeinfer leaked: {loaded}" + + +def test_make_alias_oracle_defaults_to_scalpel_without_typed_ast(): + import sys + assert "typed_ast" not in sys.modules + from codeanalyzer.dataflow.scalpel_oracle import make_alias_oracle, ScalpelAliasOracle + + src = "def f(a):\n b = a\n c = b\n return c\n" + func_ast = ast.parse(src).body[0] + oracle = make_alias_oracle(pycallable=None, func_ast=func_ast, base_types={}) + # The whole point: scalpel is the default, not the type-based fallback. + assert isinstance(oracle, ScalpelAliasOracle), type(oracle).__name__ + # It answers queries (copies alias; unrelated locals do not). + assert oracle.may_alias("b", "c") is True + + +def test_make_alias_oracle_is_deterministic(): + import ast as _ast + from codeanalyzer.dataflow.scalpel_oracle import make_alias_oracle + src = "def f(a):\n b = a\n c = b\n return c\n" + fa = _ast.parse(src).body[0] + o1 = make_alias_oracle(None, _ast.parse(src).body[0], {}) + o2 = make_alias_oracle(None, _ast.parse(src).body[0], {}) + pairs = [("a", "b"), ("b", "c"), ("a", "c"), ("b", "b")] + assert [o1.may_alias(x, y) for x, y in pairs] == [o2.may_alias(x, y) for x, y in pairs] + + +def _pip_scalpel_available(): + try: + import importlib.util + # the EXTERNAL package, not our vendored copy + return importlib.util.find_spec("scalpel") is not None and \ + not importlib.util.find_spec("scalpel").origin.endswith( + "codeanalyzer/dataflow/scalpel/__init__.py") + except Exception: + return False + + +@pytest.mark.skipif(not _pip_scalpel_available(), + reason="upstream python-scalpel not installed (3.12+ can't build typed_ast)") +def test_vendored_ssa_matches_upstream(): + """On a Python where pip python-scalpel installs (<=3.11), the vendored copy + must produce identical SSA const_dict keys as upstream on the same source.""" + import scalpel.SSA.const as up_ssa + import scalpel.cfg as up_cfg + from codeanalyzer.dataflow.scalpel.SSA.const import SSA as VSSA + from codeanalyzer.dataflow.scalpel.cfg import CFGBuilder as VCFG + + src = "def f(a):\n b = a\n if a:\n b = 2\n return b\n" + up_c = up_cfg.CFGBuilder().build_from_src("m", src) + v_c = VCFG().build_from_src("m", src) + up_fn = list(up_c.functioncfgs.values())[0] + v_fn = list(v_c.functioncfgs.values())[0] + _, up_const = up_ssa.SSA().compute_SSA(up_fn) + _, v_const = VSSA().compute_SSA(v_fn) + assert sorted(map(str, up_const)) == sorted(map(str, v_const))