Repository navigation
Expand file tree
/
Copy pathcodemodder.py
More file actions
298 lines (247 loc) · 8.88 KB
/
Copy pathcodemodder.py
File metadata and controls
298 lines (247 loc) · 8.88 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
from concurrent.futures import ThreadPoolExecutor
import datetime
import itertools
import logging
import os
import sys
from pathlib import Path
import libcst as cst
from libcst.codemod import CodemodContext
from codemodder.file_context import FileContext
from codemodder import registry, __version__
from codemodder.logging import configure_logger, logger, log_section, log_list
from codemodder.cli import parse_args
from codemodder.change import ChangeSet
from codemodder.code_directory import file_line_patterns, match_files
from codemodder.context import CodemodExecutionContext
from codemodder.diff import create_diff_from_tree
from codemodder.executor import CodemodExecutorWrapper
from codemodder.project_analysis.python_repo_manager import PythonRepoManager
from codemodder.report.codetf_reporter import report_default
from codemodder.result import ResultSet
from codemodder.semgrep import run as run_semgrep
def update_code(file_path, new_code):
"""
Write the `new_code` to the `file_path`
"""
with open(file_path, "w", encoding="utf-8") as f:
f.write(new_code)
def find_semgrep_results(
context: CodemodExecutionContext,
codemods: list[CodemodExecutorWrapper],
) -> ResultSet:
"""Run semgrep once with all configuration files from all codemods and return a set of applicable rule IDs"""
yaml_files = list(
itertools.chain.from_iterable(
[codemod.yaml_files for codemod in codemods if codemod.yaml_files]
)
)
if not yaml_files:
return ResultSet()
return run_semgrep(context, yaml_files)
def apply_codemod_to_file(
base_directory: Path,
file_context,
codemod_kls: CodemodExecutorWrapper,
source_tree,
dry_run: bool = False,
):
wrapper = cst.MetadataWrapper(source_tree)
codemod = codemod_kls(CodemodContext(wrapper=wrapper), file_context)
if not codemod.should_transform:
return False
with file_context.timer.measure("transform"):
output_tree = codemod.transform_module(source_tree)
# TODO: we can probably just use the presence of recorded changes instead of
# comparing the trees to gain some efficiency
if output_tree.deep_equals(source_tree):
return False
diff = create_diff_from_tree(source_tree, output_tree)
change_set = ChangeSet(
str(file_context.file_path.relative_to(base_directory)),
diff,
changes=file_context.codemod_changes,
)
file_context.add_result(change_set)
if not dry_run:
with file_context.timer.measure("write"):
update_code(file_context.file_path, output_tree.code)
return True
# pylint: disable-next=too-many-arguments
def process_file(
idx: int,
file_path: Path,
base_directory: Path,
codemod,
results: ResultSet,
cli_args,
) -> FileContext:
logger.debug("scanning file %s", file_path)
if idx and idx % 100 == 0:
logger.info("scanned %s files...", idx) # pragma: no cover
line_exclude = file_line_patterns(file_path, cli_args.path_exclude)
line_include = file_line_patterns(file_path, cli_args.path_include)
findings_for_rule = results.results_for_rule_and_file(
codemod.name, # TODO: should be full ID
file_path,
)
file_context = FileContext(
base_directory,
file_path,
line_exclude,
line_include,
findings_for_rule,
)
try:
with file_context.timer.measure("parse"):
with open(file_path, "r", encoding="utf-8") as f:
source_tree = cst.parse_module(f.read())
except Exception:
file_context.add_failure(file_path)
logger.exception("error parsing file %s", file_path)
return file_context
apply_codemod_to_file(
base_directory,
file_context,
codemod,
source_tree,
cli_args.dry_run,
)
return file_context
def analyze_files(
execution_context: CodemodExecutionContext,
files_to_analyze,
codemod,
results: ResultSet,
cli_args,
):
with ThreadPoolExecutor(max_workers=cli_args.max_workers) as executor:
logger.debug(
"using executor with %s threads",
cli_args.max_workers,
)
analysis_results = executor.map(
lambda args: process_file(
args[0],
args[1],
execution_context.directory,
codemod,
results,
cli_args,
),
enumerate(files_to_analyze),
)
executor.shutdown(wait=True)
execution_context.process_results(codemod.id, analysis_results)
def log_report(context, argv, elapsed_ms, files_to_analyze):
log_section("report")
logger.info("scanned: %s files", len(files_to_analyze))
all_failures = context.get_failed_files()
logger.info(
"failed: %s files (%s unique)",
len(all_failures),
len(set(all_failures)),
)
all_changes = context.get_changed_files()
logger.info(
"changed: %s files (%s unique)",
len(all_changes),
len(set(all_changes)),
)
logger.info("report file: %s", argv.output)
logger.info("total elapsed: %s ms", elapsed_ms)
logger.info(" semgrep: %s ms", context.timer.get_time_ms("semgrep"))
logger.info(" parse: %s ms", context.timer.get_time_ms("parse"))
logger.info(" transform: %s ms", context.timer.get_time_ms("transform"))
logger.info(" write: %s ms", context.timer.get_time_ms("write"))
def apply_codemods(
context: CodemodExecutionContext,
codemods_to_run: list[CodemodExecutorWrapper],
semgrep_results: ResultSet,
files_to_analyze: list[Path],
argv,
):
log_section("scanning")
if not files_to_analyze:
logger.info("no files to scan")
return
if not codemods_to_run:
logger.info("no codemods to run")
return
semgrep_finding_ids = semgrep_results.all_rule_ids()
# run codemods one at a time making sure to respect the given sequence
for codemod in codemods_to_run:
# Unfortunately the IDs from semgrep are not fully specified
# TODO: eventually we need to be able to use fully specified IDs here
if codemod.is_semgrep and codemod.name not in semgrep_finding_ids:
logger.debug(
"no results from semgrep for %s, skipping analysis",
codemod.id,
)
continue
logger.info("running codemod %s", codemod.id)
semgrep_files = semgrep_results.files_for_rule(codemod.name)
# Non-semgrep codemods ignore the semgrep results
results = codemod.apply(context, semgrep_files)
analyze_files(
context,
files_to_analyze,
codemod,
results,
argv,
)
context.process_dependencies(codemod.id)
context.log_changes(codemod.id)
def run(original_args) -> int:
start = datetime.datetime.now()
codemod_registry = registry.load_registered_codemods()
# A little awkward, but we need the codemod registry in order to validate potential arguments
argv = parse_args(original_args, codemod_registry)
if not os.path.exists(argv.directory):
logger.error(
"given directory '%s' doesn't exist or can’t be read",
argv.directory,
)
return 1
configure_logger(argv.verbose, argv.log_format, argv.project_name)
log_section("startup")
logger.info("codemodder: python/%s", __version__)
repo_manager = PythonRepoManager(Path(argv.directory))
context = CodemodExecutionContext(
Path(argv.directory),
argv.dry_run,
argv.verbose,
codemod_registry,
repo_manager,
)
repo_manager.parse_project()
# TODO: this should be a method of CodemodExecutionContext
codemods_to_run = codemod_registry.match_codemods(
argv.codemod_include, argv.codemod_exclude
)
log_section("setup")
log_list(logging.INFO, "running", codemods_to_run, predicate=lambda c: c.id)
log_list(logging.INFO, "including paths", argv.path_include)
log_list(logging.INFO, "excluding paths", argv.path_exclude)
files_to_analyze: list[Path] = match_files(
context.directory, argv.path_exclude, argv.path_include
)
full_names = [str(path) for path in files_to_analyze]
log_list(logging.DEBUG, "matched files", full_names)
semgrep_results: ResultSet = find_semgrep_results(context, codemods_to_run)
apply_codemods(
context,
codemods_to_run,
semgrep_results,
files_to_analyze,
argv,
)
results = context.compile_results(codemods_to_run)
elapsed = datetime.datetime.now() - start
elapsed_ms = int(elapsed.total_seconds() * 1000)
report_default(elapsed_ms, argv, original_args, results)
log_report(context, argv, elapsed_ms, files_to_analyze)
return 0
def main():
sys_argv = sys.argv[1:]
sys.exit(run(sys_argv))