|
| 1 | +""" |
| 2 | +Check that log.error() and add_error_log calls use constant string literals as first argument. |
| 3 | +This script scans all Python files in ddtrace/ and reports violations. |
| 4 | +Exceptions can be specified in the EXCEPTIONS set using: |
| 5 | +- "filepath:line" to exclude a specific line in a file |
| 6 | +""" |
| 7 | + |
| 8 | +import ast |
| 9 | +import pathlib |
| 10 | +import sys |
| 11 | +from typing import List |
| 12 | +from typing import Tuple |
| 13 | + |
| 14 | + |
| 15 | +# Line-specific exceptions to exclude from checking |
| 16 | +# Format: "filepath:line" to exclude a specific line in a file |
| 17 | +EXCEPTIONS = { |
| 18 | + # only constant message can be log.error() |
| 19 | + "ddtrace/internal/telemetry/logging.py:18", |
| 20 | + # log.exception calls use constant messages |
| 21 | + "ddtrace/contrib/internal/aws_lambda/patch.py:36", |
| 22 | + # log.error in _probe/registry.py ends up with a log.debug() |
| 23 | + "ddtrace/debugging/_probe/registry.py:137", |
| 24 | + "ddtrace/debugging/_probe/registry.py:146", |
| 25 | + # we added a constant check for the wrapping method of add_error_log |
| 26 | + "ddtrace/appsec/_iast/_metrics.py:53", |
| 27 | + # we added a constant check for the wrapping method of iast_error |
| 28 | + "ddtrace/appsec/_iast/_logs.py:41", |
| 29 | + "ddtrace/appsec/_iast/_logs.py:45", |
| 30 | + # the non constant part is an object type |
| 31 | + "ddtrace/appsec/_iast/_taint_tracking/_taint_objects_base.py:75", |
| 32 | +} |
| 33 | + |
| 34 | + |
| 35 | +class LogMessageChecker(ast.NodeVisitor): |
| 36 | + def __init__(self, filepath: str): |
| 37 | + self.filepath = filepath |
| 38 | + self.errors: List[Tuple[int, int]] = [] |
| 39 | + |
| 40 | + def _has_send_to_telemetry_false(self, node: ast.Call) -> bool: |
| 41 | + """Check if the call has extra={'send_to_telemetry': False}.""" |
| 42 | + for keyword in node.keywords: |
| 43 | + if keyword.arg == "extra" and isinstance(keyword.value, ast.Dict): |
| 44 | + for key, value in zip(keyword.value.keys, keyword.value.values): |
| 45 | + if ( |
| 46 | + isinstance(key, ast.Constant) |
| 47 | + and key.value == "send_to_telemetry" |
| 48 | + and isinstance(value, ast.Constant) |
| 49 | + and value.value is False |
| 50 | + ): |
| 51 | + return True |
| 52 | + return False |
| 53 | + |
| 54 | + def visit_Call(self, node: ast.Call) -> None: |
| 55 | + """Check if this is a log.error(), add_error_log, or iast_error call with non-constant first arg.""" |
| 56 | + fn = node.func |
| 57 | + |
| 58 | + # Check for add_error_log calls |
| 59 | + is_add_integration_error = isinstance(fn, ast.Attribute) and fn.attr == "add_error_log" |
| 60 | + # Check for log.error() calls (simple check for .error() on any variable) |
| 61 | + is_log_error = isinstance(fn, ast.Attribute) and (fn.attr == "error" or fn.attr == "exception") |
| 62 | + # Check for iast_error calls |
| 63 | + is_iast_log = isinstance(fn, ast.Name) and ( |
| 64 | + fn.id == "iast_error" |
| 65 | + or fn.id == "iast_instrumentation_ast_patching_errorr_log" |
| 66 | + or fn.id == "iast_propagation_error_log" |
| 67 | + ) |
| 68 | + is_target = is_add_integration_error or is_log_error or is_iast_log |
| 69 | + |
| 70 | + if is_target and node.args: |
| 71 | + msg = node.args[0] |
| 72 | + is_constant_string = isinstance(msg, ast.Constant) and isinstance(msg.value, str) |
| 73 | + |
| 74 | + # Skip constant string check if send_to_telemetry is False for log.error/exception calls |
| 75 | + if not is_constant_string and is_log_error and self._has_send_to_telemetry_false(node): |
| 76 | + pass |
| 77 | + elif not is_constant_string and not self._is_line_exception(node.lineno): |
| 78 | + self.errors.append((node.lineno, node.col_offset)) |
| 79 | + |
| 80 | + self.generic_visit(node) |
| 81 | + |
| 82 | + def _is_line_exception(self, line_no: int) -> bool: |
| 83 | + """Check if this specific line is in the exceptions list.""" |
| 84 | + return f"{str(self.filepath)}:{line_no}" in EXCEPTIONS |
| 85 | + |
| 86 | + |
| 87 | +def check_file(filepath: pathlib.Path) -> List[Tuple[int, int]]: |
| 88 | + try: |
| 89 | + source = filepath.read_text(encoding="utf-8") |
| 90 | + tree = ast.parse(source, filename=str(filepath)) |
| 91 | + checker = LogMessageChecker(str(filepath)) |
| 92 | + checker.visit(tree) |
| 93 | + return checker.errors |
| 94 | + except (OSError, UnicodeDecodeError) as e: |
| 95 | + print(f"Error reading {filepath}: {e}", file=sys.stderr) |
| 96 | + return [] |
| 97 | + except SyntaxError as e: |
| 98 | + print(f"Syntax error in {filepath}:{e.lineno}:{e.offset}: {e.msg}", file=sys.stderr) |
| 99 | + return [] |
| 100 | + |
| 101 | + |
| 102 | +def main() -> int: |
| 103 | + contrib_path = pathlib.Path("ddtrace") |
| 104 | + python_files = list(contrib_path.rglob("*.py")) |
| 105 | + |
| 106 | + total_errors = 0 |
| 107 | + |
| 108 | + for filepath in python_files: |
| 109 | + errors = check_file(filepath) |
| 110 | + for line_no, col_no in errors: |
| 111 | + print(f"{filepath}:{line_no}:{col_no}: " "LOG001 first argument to logging call must be a constant string") |
| 112 | + total_errors += 1 |
| 113 | + |
| 114 | + if total_errors > 0: |
| 115 | + print(f"\nFound {total_errors} violation(s)", file=sys.stderr) |
| 116 | + return 1 |
| 117 | + |
| 118 | + print("All logging calls use constant strings ✓") |
| 119 | + return 0 |
| 120 | + |
| 121 | + |
| 122 | +if __name__ == "__main__": |
| 123 | + sys.exit(main()) |
0 commit comments