#!/usr/bin/env python3 # Copyright 2026 The Chromium Authors # Use of this source code is governed by a BSD-style license that can be # found in the LICENSE file. """ A tool to identify and remove unnecessary UNSAFE_TODO() wrappers in Chromium. This script automates the verification process by locating all instances of UNSAFE_TODO(), determining if the wrapper is redundant, and confirming that the code is compiled in the current build configuration. Because this process requires approximately 10k-40k compiler invocations, we can't afford the latency of siso/remoteexec/autoninja for each. Thus, this tool compiles locally a single file from the compile_commands.json database generated by GN. This works because UNSAFE_TODO() is a compile-time check. The only caveat are UNSAFE_TODO in templates instantiations, which are not verified by this method. The script supports state persistence via a JSON file, enabling the analysis to resume without losing progress. It also supports multiple builders, allowing verification of the same UNSAFE_TODO() across different configurations. Usage: ./unsafe_todo_remover.py # Run analysis ./unsafe_todo_remover.py --apply # Apply fixes from out.json """ from __future__ import annotations import argparse import json import os import re import shlex import subprocess import sys from collections import defaultdict from collections.abc import Sequence # --- Configuration --- # State is persisted in this file. This allows resuming the analysis # without losing progress. STATE_FILE = "out.json" # A set of builders covering most of the chromium C++ code. # # See infra/config/generated/builders/gn_args_locations.json for reference. BUILDERS = [ # Linux builders ('tryserver.chromium.linux', 'linux-rel'), ('tryserver.chromium.linux', 'linux_chromium_tsan_rel_ng'), ('chromium.linux', 'Linux Builder (dbg)'), ('chromium.memory', 'Linux ASan LSan Builder'), # Chromeos builders ('chromium.chromiumos', 'linux-cfm-rel'), ('chromium.chromiumos', 'linux-chromeos-rel'), ('tryserver.chromium.chromiumos', 'chromeos-libfuzzer-asan-rel'), # Android builders ('chromium.android', 'android-x86-rel'), ('chromium', 'android-arm64-archive-rel'), ('chromium.android', 'android-14-x64-rel'), ('chromium.android', 'android-x86-rel'), ('chromium.fuzz', 'android-arm64-libfuzzer-hwasan'), ('tryserver.chromium.android', 'android-cronet-arm-dbg'), ('tryserver.chromium.android', 'android-webview-13-x64-dbg'), # Fuchsia ('chromium.fuchsia', 'Deterministic Fuchsia (dbg)'), # Mac ('chromium.mac', 'mac-arm64-dbg'), ('chromium.mac', 'mac-arm64-rel'), ('tryserver.chromium.mac', 'mac-rel'), ('tryserver.chromium.mac', 'mac-rel'), # Dawn ('tryserver.chromium.dawn', 'dawn-win11-arm64-deps-rel'), # Windows builders ('chromium.win', 'linux-win-cross-rel'), ('chromium.memory', 'win-asan'), ('chromium.win', 'Win Builder'), ('chromium.win', 'Win x64 Builder (dbg)'), ('chromium.win', 'linux-win-cross-rel'), ('chromium.win', 'win-arm64-rel'), ('tryserver.chromium.win', 'win-arm64-compile-dbg'), ('tryserver.chromium.win', 'win-libfuzzer-asan-rel'), ('tryserver.chromium.win', 'win-rel'), ('tryserver.chromium.win', 'win_optional_gpu_tests_rel'), # Fuchsia cast ('chromium.fuchsia', 'fuchsia-x64-cast-receiver-rel'), # Android cast ('chromium.android', "android-cast-arm-dbg"), ] EXCLUDED_DIRECTORIES = [ # Non production code: "agents/", "tools/", "build/", # UNSAFE_TODO in template. # # C++ templates can explode into many instantiations, making it impractical # to verify all of them. This script only build the file that contains the # UNSAFE_TODO, so template instantiations in other files are not verified. # Thus, we exclude files known to have UNSAFE_TODO in templates. "base/strings/string_tokenizer.h", "chrome/browser/media/router/discovery/discovery_network_list_posix.cc", "components/zucchini/io_utils.h", "components/zucchini/patch_utils.h", "components/zucchini/rel32_utils.h", "device/fido/cbor_extract.h", "gpu/command_buffer/common/cmd_buffer_common.h", "gpu/command_buffer/common/gles2_cmd_utils.h", "mojo/core/broker_messages.h", "mojo/public/cpp/bindings/lib/array_internal.h", "mojo/public/cpp/bindings/lib/array_serialization.h", "third_party/blink/renderer/core/css/parser/css_tokenizer_input_stream.h", "third_party/blink/renderer/platform/fonts/opentype/open_type_types.h", "ui/gfx/x/property_cache.h", # UNSAFE_TODO wrapping specific macros. # # Those macros expands sometimes into code that triggers # -WUnsafe-buffer-usage or not, depending on the platform. These variations # are not supported by the script, so we exclude them. "chrome/browser/media/router/discovery/discovery_network_list_posix.cc", "net/base/network_interfaces_posix.cc", "net/dns/address_info.cc", "net/dns/loopback_only.cc", "media/gpu/test/video_test_helpers.cc", "net/cert/ct_objects_extractor.cc", "sandbox/linux/syscall_broker/broker_simple_message.cc", ] # List of functions that are considered unsafe and should not be unwrapped. # There is not need to check the content of UNSAFE_TODO for these functions, # as they are always unsafe. This is used to mark them as LIBC in the state # file, which means they cannot be removed. # This list is based on the Clang source code. UNSAFE_LIBC_FUNCTIONS = { "atof", "atoi", "atol", "atoll", "bcopy", "bsearch", "bzero", "fgets", "fgetws", "fputs", "fputws", "fread", "fwrite", "gets", "memccpy", "memchr", "memcmp", "memcpy", "memmove", "mempcpy", "memset", "puts", "qsort", "strcasecmp", "strcat", "strchr", "strcmp", "strcoll", "strcpy", "strcspn", "strdup", "strerror_r", "strerror_s", "stricmp", "strlcat", "strlcpy", "strlen", "strncat", "strncmp", "strncpy", "strndup", "strnlen", "strpbrk", "strrchr", "strspn", "strstr", "strtod", "strtof", "strtoimax", "strtok", "strtol", "strtold", "strtoll", "strtoul", "strtoull", "strtoumax", "strxfrm", "wmemchr", "wmemcmp", "wmemcpy", "wmemmove", "wmemset", } # Regex to match: optional scope (:: or std::), one of the functions, followed # by word boundary. Used to check the content inside UNSAFE_TODO(...). LIBC_REGEX = re.compile( r'^\s*(?:(?:std)?::)?(?:' + '|'.join(re.escape(f) for f in UNSAFE_LIBC_FUNCTIONS) + r')\b' ) # --- Helpers --- class Colors: CYAN = '\033[96m' GREEN = '\033[92m' YELLOW = '\033[93m' RED = '\033[91m' MAGENTA = '\033[95m' BLUE = '\033[94m' RESET = '\033[0m' def log(category: str, message: str): """Prints a categorized, colored message to stderr.""" cat_upper = category.upper() color = Colors.RESET if cat_upper in ("ERR", "FAIL", "ERROR"): color = Colors.RED elif cat_upper in ("WARN", "WARNING"): color = Colors.YELLOW elif cat_upper in ("OK", "SUCCESS", "PASS"): color = Colors.GREEN elif cat_upper in ("INFO", "SCAN", "EDIT", "FMT") or cat_upper.isdigit(): color = Colors.CYAN elif cat_upper in ("BUILD", "GN", "BUILDER"): color = Colors.MAGENTA print(f"{color}[{cat_upper}]{Colors.RESET} {message}", file=sys.stderr) def run_command( cmd: Sequence[str], cwd: str | None = None, capture_output: bool = True ) -> subprocess.CompletedProcess: """Executes a subprocess command.""" try: return subprocess.run( cmd, cwd=cwd, text=True, capture_output=capture_output, check=False ) except Exception as e: log("ERR", f"Command execution failed: {cmd} -> {e}") return subprocess.CompletedProcess(cmd, 1, "", str(e)) def is_excluded(path: str) -> bool: """Checks if a path should be ignored based on project rules.""" # 1. Only modify C++ files valid_extensions = ('.cc', '.h', '.cpp', '.mm', '.hh') if not path.endswith(valid_extensions): return True # 2. Specific directory exclusions if path.startswith(tuple(EXCLUDED_DIRECTORIES)): return True # 3. Third party logic: Exclude all third_party EXCEPT third_party/blink if path.startswith("third_party/"): if not path.startswith("third_party/blink/"): return True return False # --- Core Logic --- class StateManager: """Manages the persistence of analysis results.""" def __init__(self, filepath: str): self.filepath = filepath # Structure: { file_key: { builder_name: status } } self.data: dict[str, dict[str, str]] = {} self.load() def load(self): if os.path.exists(self.filepath): try: with open(self.filepath, 'r') as f: self.data = json.load(f) except json.JSONDecodeError: log("WARN", f"Corrupt {self.filepath}, starting fresh.") def save(self): with open(self.filepath, 'w') as f: json.dump(self.data, f, indent=2, sort_keys=True) def get_builder_status(self, key: str, builder: str) -> str | None: """Returns the status of a specific builder for the given key.""" entry = self.data.get(key, {}) # Handle legacy format where value was a string if isinstance(entry, str): return None return entry.get(builder) def set_status(self, key: str, builder: str, status: str): """Sets the status for a specific builder.""" if key not in self.data or not isinstance(self.data[key], dict): self.data[key] = {} self.data[key][builder] = status self.save() def is_resolved(self, key: str) -> bool: """ Returns True if the UNSAFE_TODO has a definitive result (SUCCESS or FAIL) from *any* builder, or is marked as LIBC. """ if key not in self.data: return False entry = self.data[key] if isinstance(entry, str): # Legacy format support return entry in ["SUCCESS", "FAIL", "LIBC"] vals = entry.values() return "SUCCESS" in vals or "FAIL" in vals or "LIBC" in vals class SourceLocation: """Represents a specific UNSAFE_TODO instance.""" def __init__(self, filepath: str, line: int, col: int, content: str): self.filepath = filepath self.line = line self.col = col self.original_content = content # Unique key for state tracking: "path/to/file.cc:10:5" self.key = f"{filepath}:{line}:{col}" def __repr__(self): return f"{self.filepath}:{self.line}" class BuildEnvironment: """Manages the GN/Ninja build environment.""" def __init__(self, builder_group: str, builder_config: str): self.group = builder_group self.config = builder_config # Sanitize directory name to avoid spaces/parens. # Allow alphanumeric, underscore, dash. Replace others with underscore. safe_dirname = re.sub(r'[^a-zA-Z0-9_-]', '_', builder_config) self.build_dir = os.path.join("out", safe_dirname) self.is_ready = False self.compile_db: dict[str, str] = {} def ensure_ready(self): """Sets up the build directory if it doesn't exist.""" if self.is_ready: return if not os.path.exists(self.build_dir): log("BUILD", f"Setting up builder: {self.config}...") cmd = [ "./tools/mb/mb.py", "gen", "-m", self.group, "-b", self.config, self.build_dir, ] res = run_command(cmd) if res.returncode != 0: log( "ERR", f"Failed to generate build files for {self.config}: {res.stderr}", ) raise RuntimeError("MB gen failed") # Always regenerate compile_commands.json to ensure it's fresh log("GN", f"Generating compile_commands.json for {self.config}...") res = run_command( ["gn", "gen", self.build_dir, "--export-compile-commands"] ) if res.returncode != 0: log("ERR", f"Failed to export compile commands: {res.stderr}") raise RuntimeError("GN gen failed") # Build everything to ensure generated files (headers, etc.) are present. log( "BUILD", f"Building all targets in {self.config} (this may take a while)...", ) run_command(["autoninja", "-C", self.build_dir], capture_output=False) self._load_compile_db() self.is_ready = True def _load_compile_db(self): """Loads the compile_commands.json database.""" db_path = os.path.join(self.build_dir, "compile_commands.json") try: with open(db_path, 'r') as f: data = json.load(f) self.compile_db = {} for entry in data: directory = entry.get('directory', '.') filepath = entry.get('file', '') if not filepath: continue abs_path = os.path.abspath(os.path.join(directory, filepath)) self.compile_db[abs_path] = entry.get('command', '') except (IOError, json.JSONDecodeError) as e: log("WARN", f"Failed to load compile_commands.json: {e}") def has_file(self, filepath: str) -> bool: """Checks if the file exists in the compile database.""" abs_path = os.path.abspath(filepath) return abs_path in self.compile_db def find_compilable_source(self, filepath: str) -> str | None: """Returns a file that can be compiled for the given filepath.""" abs_path = os.path.abspath(filepath) if abs_path in self.compile_db: return filepath # Header fallback (Source/Header proxy) if filepath.endswith(('.h', '.hh', '.hpp')): base = os.path.splitext(filepath)[0] for ext in ['.cc', '.cpp', '.mm']: surrogate = base + ext abs_surrogate = os.path.abspath(surrogate) if abs_surrogate in self.compile_db: return surrogate return None def compile_file(self, filepath: str) -> bool: """ Compiles a single file using the `compile_commands.json` database. """ compilable_source = self.find_compilable_source(filepath) if not compilable_source: return False cmd_list = shlex.split(self.compile_db[os.path.abspath(compilable_source)]) res = run_command(cmd_list, cwd=self.build_dir, capture_output=False) return res.returncode == 0 class CodeModifier: """Handles parsing and modifying source files.""" @staticmethod def find_unsafe_todos() -> Sequence[SourceLocation]: """Scans the codebase for UNSAFE_TODO macros.""" log("SCAN", "Scanning for UNSAFE_TODO instances...") cmd = [ "git", "grep", "-n", "--column", "-w", "UNSAFE_TODO", "--", "*.cc", "*.h", "*.cpp", "*.mm", "*.hh", ] res = run_command(cmd) lines = res.stdout.strip().splitlines() locations = [] for line in lines: parts = line.split(":", 3) # path:line:col:content if len(parts) < 4: continue fpath = parts[0] linenum = int(parts[1]) colnum = int(parts[2]) if is_excluded(fpath): continue locations.append(SourceLocation(fpath, linenum, colnum, "")) return locations @staticmethod def _read_file(filepath: str) -> Sequence[str]: with open(filepath, 'r') as f: return f.readlines() @staticmethod def _write_file(filepath: str, lines: Sequence[str]): with open(filepath, 'w') as f: f.writelines(lines) @staticmethod def replace_macro(loc: SourceLocation): """ Simple string replacement on the specific line. We replace UNSAFE_TODO with UNSAFE_XXXX. """ lines = CodeModifier._read_file(loc.filepath) line_idx = loc.line - 1 col_idx = loc.col - 1 original_line = lines[line_idx] lines[line_idx] = ( original_line[:col_idx] + "UNSAFE_XXXX" + original_line[col_idx + len("UNSAFE_TODO") :] ) CodeModifier._write_file(loc.filepath, lines) @staticmethod def _remove_wrapper_from_content(content: str, line: int, col: int) -> str: """ Internal logic to remove one wrapper from string content. Note: Relying on parenthesis matching is "good" enough here, most of the time. Failures will simply skip the removal. """ # Find the line start offset lines = content.splitlines(keepends=True) offset = 0 for i in range(line - 1): offset += len(lines[i]) offset += col - 1 # Verify we are at UNSAFE_TODO if not content[offset:].startswith("UNSAFE_TODO"): log("WARN", f"Offset mismatch at {line}:{col}. Skipping...") return content # Find opening paren open_paren = content.find("(", offset) if open_paren == -1: return content # Find matching closing paren balance = 1 close_paren = -1 for i in range(open_paren + 1, len(content)): if content[i] == "(": balance += 1 elif content[i] == ")": balance -= 1 if balance == 0: close_paren = i break if close_paren == -1: return content # Extract inner content inner = content[open_paren + 1 : close_paren] # Reconstruct new_content = content[:offset] + inner + content[close_paren + 1 :] return new_content @staticmethod def remove_wrapper(loc: SourceLocation): """Removes UNSAFE_TODO(...) wrapper, keeping inner content.""" try: with open(loc.filepath, 'r') as f: content = f.read() new_content = CodeModifier._remove_wrapper_from_content( content, loc.line, loc.col ) if content != new_content: with open(loc.filepath, 'w') as f: f.write(new_content) except Exception as e: log("ERR", f"Failed to modify {loc}: {e}") @staticmethod def is_libc_call(loc: SourceLocation) -> bool: """Checks if the UNSAFE_TODO wraps a known unsafe libc function.""" try: with open(loc.filepath, 'r') as f: content = f.read() # Find start offset based on line/col lines = content.splitlines(keepends=True) offset = 0 for i in range(loc.line - 1): offset += len(lines[i]) offset += loc.col - 1 if not content[offset:].startswith("UNSAFE_TODO"): return False open_paren = content.find("(", offset) if open_paren == -1: return False balance = 1 close_paren = -1 for i in range(open_paren + 1, len(content)): if content[i] == "(": balance += 1 elif content[i] == ")": balance -= 1 if balance == 0: close_paren = i break if close_paren == -1: return False inner = content[open_paren + 1 : close_paren].strip() return bool(LIBC_REGEX.match(inner)) except Exception: return False @staticmethod def remove_wrappers(locs: Sequence[SourceLocation]): """Removes multiple wrappers from the same file efficiently.""" if not locs: return filepath = locs[0].filepath try: with open(filepath, 'r') as f: content = f.read() # Sort in reverse order (bottom-up, right-to-left) to preserve offsets # of earlier modifications. sorted_locs = sorted(locs, key=lambda x: (x.line, x.col), reverse=True) original_content = content for loc in sorted_locs: content = CodeModifier._remove_wrapper_from_content( content, loc.line, loc.col ) if content != original_content: with open(filepath, 'w') as f: f.write(content) except Exception as e: log("ERR", f"Failed to batch modify {filepath}: {e}") @staticmethod def revert_file(filepath: str): """Restores file using git checkout.""" run_command(["git", "checkout", filepath]) # --- Main Analysis Loop --- def analyze_todos(args): state = StateManager(STATE_FILE) todos = CodeModifier.find_unsafe_todos() log("INFO", f"Found {len(todos)} UNSAFE_TODO candidates.") # Pre-scan for LIBC functions log("INFO", "Checking for known LIBC functions...") libc_count = 0 for todo in todos: if state.is_resolved(todo.key): continue if CodeModifier.is_libc_call(todo): state.set_status(todo.key, "scanner", "LIBC") libc_count += 1 if libc_count > 0: log("INFO", f"Marked {libc_count} UNSAFE_TODOs as LIBC (non-removable).") # Group TODOs by file todos_by_file = defaultdict(list) for todo in todos: if not state.is_resolved(todo.key): todos_by_file[todo.filepath].append(todo) log("INFO", f"{len(todos_by_file)} files remain unresolved.") # Initialize builders builders = [BuildEnvironment(g, c) for g, c in BUILDERS] for builder in builders: try: builder.ensure_ready() except Exception as e: log("ERR", f"Skipping builder {builder.config} due to setup failure: {e}") continue log("BUILDER", f"Processing files on builder: {builder.config}...") processed_files_count = 0 filenames = list(todos_by_file.keys()) for filepath in filenames: file_todos = todos_by_file[filepath] # 1. Check if the file is already resolved unresolved_todos = [t for t in file_todos if not state.is_resolved(t.key)] if not unresolved_todos: continue # 2. Check if this builder has already run for these TODOs todos_to_check = [ t for t in unresolved_todos if not state.get_builder_status(t.key, builder.config) ] if not todos_to_check: continue processed_files_count += 1 log( str(processed_files_count), f"Processing {filepath} ({len(todos_to_check)} items)...", ) try: # --- STEP 0: Skip files that can't be compiled --- if not builder.find_compilable_source(filepath): log( "WARN", f"Cannot compile {filepath} on {builder.config}. Skipping." ) for todo in todos_to_check: state.set_status(todo.key, builder.config, "SKIP") continue # --- STEP 1: Individual Checks --- for todo in todos_to_check: # Inverted Check Logic: # 1. First, check if removing the wrapper compile. # 2. Second, check if the line is active under this builder. # 1. Remove wrapper CodeModifier.remove_wrapper(todo) safety_pass = builder.compile_file(filepath) CodeModifier.revert_file(filepath) if not safety_pass: # Removing the wrapper broke the build. It is required. log("FAIL", f"Line {todo.line}: Unsafe (wrapper required).") state.set_status(todo.key, builder.config, "FAIL") continue # 2. Rename UNSAFE_TODO -> UNSAFE_XXXX to verify usage. CodeModifier.replace_macro(todo) usage_pass = builder.compile_file(filepath) CodeModifier.revert_file(filepath) if usage_pass: # UNSAFE_XXXX compiled successfully -> Macro content is dead code on # this builder (compiler didn't see error). We record SKIP so other # builders can try to verify it. state.set_status(todo.key, builder.config, "SKIP") else: # UNSAFE_XXXX failed to compile -> Code is live. Since safety_pass # was True (removal worked), this is a confirmed Success. log("OK", f"Line {todo.line}: Safe.") state.set_status(todo.key, builder.config, "SUCCESS") except KeyboardInterrupt: log("WARN", f"Interrupted. Reverting {filepath}...") CodeModifier.revert_file(filepath) sys.exit(1) except Exception as e: log("ERR", f"Unexpected error processing {filepath}: {e}") CodeModifier.revert_file(filepath) def apply_fixes(args): """Reads out.json and applies changes for SUCCESS entries.""" state = StateManager(STATE_FILE) if not state.data: log("WARN", "No state file found or file empty. Run analysis first.") return # Group by file to minimize IO and handle multiple edits per file files_to_edit = {} # Iterate over all keys in state for key, builder_status_map in state.data.items(): # Check for legacy format if not isinstance(builder_status_map, dict): status = builder_status_map is_success = status == "SUCCESS" else: # Check if ANY builder succeeded is_success = "SUCCESS" in builder_status_map.values() if not is_success: continue parts = key.rsplit(":", 2) fpath = parts[0] line = int(parts[1]) col = int(parts[2]) if fpath not in files_to_edit: files_to_edit[fpath] = [] files_to_edit[fpath].append(SourceLocation(fpath, line, col, "")) log("INFO", f"Applying fixes to {len(files_to_edit)} files...") for fpath, locs in files_to_edit.items(): log("EDIT", f"Updating {fpath} ({len(locs)} fixes)...") CodeModifier.remove_wrappers(locs) # Run git cl format on modified files if files_to_edit: log("FMT", "Running git cl format...") run_command(["git", "cl", "format"] + list(files_to_edit.keys())) def main(): parser = argparse.ArgumentParser(description="Cleanup UNSAFE_TODO macros") parser.add_argument( "--apply", action="store_true", help="Apply fixes based on out.json" ) args = parser.parse_args() if args.apply: apply_fixes(args) else: analyze_todos(args) if __name__ == "__main__": main()