mirror of
https://github.com/norandom/OpenUltraSAST.git
synced 2026-11-13 17:58:18 +00:00
feat(detection-core): add syntax-only migration boundary
This commit is contained in:
Marius Ciepluch
@@ -0,0 +1,453 @@
|
||||
"""Syntax-only cutover boundary for the detection-core v1 migration."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import ast
|
||||
from collections.abc import Sequence
|
||||
from dataclasses import dataclass
|
||||
from enum import Enum
|
||||
from pathlib import Path
|
||||
|
||||
from core_migration_inventory import (
|
||||
LegacyInventoryEntry,
|
||||
LegacyMigrationInventory,
|
||||
MigrationDisposition,
|
||||
validate_inventory_transition,
|
||||
)
|
||||
from core_migration_scope import (
|
||||
COMPATIBILITY_PATHS,
|
||||
COMPATIBILITY_SHIMS,
|
||||
FROZEN_TOP_LEVEL_V1,
|
||||
REPO,
|
||||
MigrationScopeError,
|
||||
authoritative_cutover_scope,
|
||||
boundary_python,
|
||||
cutover_residue_error,
|
||||
cutover_scope_error,
|
||||
expected_boundary_scope,
|
||||
partition_boundary_sites,
|
||||
relative_path,
|
||||
scope_identity,
|
||||
)
|
||||
from core_migration_scope import (
|
||||
static_top_level_v1 as _static_top_level_v1,
|
||||
)
|
||||
|
||||
|
||||
class LegacyBoundaryError(ValueError):
|
||||
"""The structural migration boundary is unreadable or not cutover-ready."""
|
||||
|
||||
|
||||
class BoundaryPhase(Enum):
|
||||
MIGRATION = "migration"
|
||||
POST_DELETION = "post_deletion"
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class LegacyBoundaryReport:
|
||||
phase: BoundaryPhase
|
||||
files_read: int
|
||||
bytes_read: int
|
||||
scanned_scope_paths: tuple[str, ...]
|
||||
scanned_scope_digest: str
|
||||
expected_scope_paths: tuple[str, ...]
|
||||
expected_scope_digest: str
|
||||
open_inventory_entries: tuple[LegacyInventoryEntry, ...]
|
||||
legacy_import_sites: tuple[str, ...]
|
||||
legacy_writer_sites: tuple[str, ...]
|
||||
forbidden_dynamic_sites: tuple[str, ...]
|
||||
consumer_legacy_import_sites: tuple[str, ...]
|
||||
consumer_legacy_writer_sites: tuple[str, ...]
|
||||
compatibility_implementation_sites: tuple[str, ...]
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class _Bindings:
|
||||
core_modules: frozenset[str]
|
||||
compatibility_modules: frozenset[str]
|
||||
writer_modules: frozenset[str]
|
||||
writer_members: frozenset[str]
|
||||
import_modules: frozenset[str]
|
||||
import_functions: frozenset[str]
|
||||
|
||||
|
||||
def static_top_level_v1(root: Path = REPO) -> tuple[str, ...]:
|
||||
try:
|
||||
return _static_top_level_v1(root)
|
||||
except MigrationScopeError as exc:
|
||||
raise LegacyBoundaryError(str(exc)) from exc
|
||||
|
||||
|
||||
def _dotted(node: ast.AST) -> str:
|
||||
if isinstance(node, ast.Name):
|
||||
return node.id
|
||||
if isinstance(node, ast.Attribute):
|
||||
prefix = _dotted(node.value)
|
||||
return f"{prefix}.{node.attr}" if prefix else ""
|
||||
return ""
|
||||
|
||||
|
||||
def _parents(tree: ast.AST) -> dict[ast.AST, ast.AST]:
|
||||
return {child: parent for parent in ast.walk(tree) for child in ast.iter_child_nodes(parent)}
|
||||
|
||||
|
||||
def _symbol(node: ast.AST, parents: dict[ast.AST, ast.AST]) -> str:
|
||||
names = []
|
||||
current = node
|
||||
while current in parents:
|
||||
current = parents[current]
|
||||
if isinstance(current, (ast.ClassDef, ast.FunctionDef, ast.AsyncFunctionDef)):
|
||||
names.append(current.name)
|
||||
return ".".join(reversed(names)) or "<module>"
|
||||
|
||||
|
||||
def _site(relative: str, node: ast.AST, parents: dict[ast.AST, ast.AST], kind: str, detail: str = "") -> str:
|
||||
suffix = f":{detail}" if detail else ""
|
||||
return f"{relative}:{getattr(node, 'lineno', 0)}:{_symbol(node, parents)}:{kind}{suffix}"
|
||||
|
||||
|
||||
def _bindings(tree: ast.Module) -> _Bindings:
|
||||
core_modules = {"openultrasast.core"}
|
||||
compatibility_modules = {"openultrasast.core.legacy_v1"}
|
||||
writer_modules = {"openultrasast.core.legacy_v1.writer"}
|
||||
writer_members: set[str] = set()
|
||||
import_modules = {"importlib"}
|
||||
import_functions = {"__import__"}
|
||||
for node in ast.walk(tree):
|
||||
if isinstance(node, ast.Import):
|
||||
for alias in node.names:
|
||||
if alias.name == "openultrasast.core":
|
||||
core_modules.add(alias.asname or "openultrasast.core")
|
||||
elif alias.name == "openultrasast.core.legacy_v1":
|
||||
compatibility_modules.add(alias.asname or alias.name)
|
||||
elif alias.name == "openultrasast.core.legacy_v1.writer":
|
||||
writer_modules.add(alias.asname or alias.name)
|
||||
elif alias.name == "importlib":
|
||||
import_modules.add(alias.asname or "importlib")
|
||||
elif isinstance(node, ast.ImportFrom):
|
||||
_update_from_bindings(node, core_modules, compatibility_modules, writer_modules, writer_members, import_functions)
|
||||
return _Bindings(
|
||||
frozenset(core_modules),
|
||||
frozenset(compatibility_modules),
|
||||
frozenset(writer_modules),
|
||||
frozenset(writer_members),
|
||||
frozenset(import_modules),
|
||||
frozenset(import_functions),
|
||||
)
|
||||
|
||||
|
||||
def _update_from_bindings(
|
||||
node: ast.ImportFrom,
|
||||
core_modules: set[str],
|
||||
compatibility_modules: set[str],
|
||||
writer_modules: set[str],
|
||||
writer_members: set[str],
|
||||
import_functions: set[str],
|
||||
) -> None:
|
||||
if node.module == "openultrasast":
|
||||
core_modules.update(alias.asname or alias.name for alias in node.names if alias.name == "core")
|
||||
elif node.module == "openultrasast.core":
|
||||
compatibility_modules.update(alias.asname or alias.name for alias in node.names if alias.name == "legacy_v1")
|
||||
elif node.module == "openultrasast.core.legacy_v1":
|
||||
writer_modules.update(alias.asname or alias.name for alias in node.names if alias.name == "writer")
|
||||
elif node.module == "openultrasast.core.legacy_v1.writer":
|
||||
writer_members.update(alias.asname or alias.name for alias in node.names)
|
||||
elif node.module == "importlib":
|
||||
import_functions.update(alias.asname or alias.name for alias in node.names if alias.name == "import_module")
|
||||
|
||||
|
||||
def _import_sites(relative: str, tree: ast.Module, parents: dict[ast.AST, ast.AST]) -> tuple[set[str], set[str], set[str]]:
|
||||
legacy: set[str] = set()
|
||||
writers: set[str] = set()
|
||||
dynamic: set[str] = set()
|
||||
for node in ast.walk(tree):
|
||||
if isinstance(node, ast.Import):
|
||||
for alias in node.names:
|
||||
_classify_import(relative, node, alias.name, parents, legacy, writers, dynamic)
|
||||
elif isinstance(node, ast.ImportFrom):
|
||||
_classify_import_from(relative, node, parents, legacy, writers, dynamic)
|
||||
return legacy, writers, dynamic
|
||||
|
||||
|
||||
def _classify_import(
|
||||
relative: str,
|
||||
node: ast.Import,
|
||||
module: str,
|
||||
parents: dict[ast.AST, ast.AST],
|
||||
legacy: set[str],
|
||||
writers: set[str],
|
||||
dynamic: set[str],
|
||||
) -> None:
|
||||
if module == "openultrasast.core":
|
||||
dynamic.add(_site(relative, node, parents, "module_valued_core_import", module))
|
||||
if module.startswith("openultrasast.core.legacy_v1"):
|
||||
legacy.add(_site(relative, node, parents, "legacy_namespace_import", module))
|
||||
if module == "openultrasast.core.legacy_v1.writer":
|
||||
writers.add(_site(relative, node, parents, "writer_facade_import", module))
|
||||
if module in COMPATIBILITY_SHIMS:
|
||||
legacy.add(_site(relative, node, parents, "compatibility_shim_import", module))
|
||||
|
||||
|
||||
def _classify_import_from(
|
||||
relative: str,
|
||||
node: ast.ImportFrom,
|
||||
parents: dict[ast.AST, ast.AST],
|
||||
legacy: set[str],
|
||||
writers: set[str],
|
||||
dynamic: set[str],
|
||||
) -> None:
|
||||
module = node.module or ""
|
||||
names = {alias.name for alias in node.names}
|
||||
if module == "openultrasast" and "core" in names:
|
||||
dynamic.add(_site(relative, node, parents, "module_valued_core_import", "openultrasast.core"))
|
||||
if module.startswith("openultrasast.core.legacy_v1") or (module == "openultrasast.core" and "legacy_v1" in names):
|
||||
legacy.add(_site(relative, node, parents, "legacy_namespace_import", module or "openultrasast.core.legacy_v1"))
|
||||
if module == "openultrasast.core.legacy_v1.writer" or (module == "openultrasast.core.legacy_v1" and "writer" in names):
|
||||
writers.add(_site(relative, node, parents, "writer_facade_import", module))
|
||||
if module == "openultrasast.core":
|
||||
if "*" in names:
|
||||
dynamic.add(_site(relative, node, parents, "top_level_star_import", module))
|
||||
for name in sorted(names.intersection(FROZEN_TOP_LEVEL_V1)):
|
||||
legacy.add(_site(relative, node, parents, "top_level_v1_import", name))
|
||||
for name in sorted(names):
|
||||
shim = f"openultrasast.core.{name}"
|
||||
if shim in COMPATIBILITY_SHIMS:
|
||||
legacy.add(_site(relative, node, parents, "compatibility_shim_import", shim))
|
||||
if module in COMPATIBILITY_SHIMS:
|
||||
legacy.add(_site(relative, node, parents, "compatibility_shim_import", module))
|
||||
|
||||
|
||||
def _reference_sites(
|
||||
relative: str,
|
||||
tree: ast.Module,
|
||||
parents: dict[ast.AST, ast.AST],
|
||||
bindings: _Bindings,
|
||||
) -> tuple[set[str], set[str], set[str]]:
|
||||
legacy: set[str] = set()
|
||||
writers: set[str] = set()
|
||||
dynamic: set[str] = set()
|
||||
for node in ast.walk(tree):
|
||||
if isinstance(node, ast.Attribute):
|
||||
_classify_attribute(relative, node, parents, bindings, legacy, writers)
|
||||
elif isinstance(node, ast.Call):
|
||||
_classify_call(relative, node, parents, bindings, writers, dynamic)
|
||||
return legacy, writers, dynamic
|
||||
|
||||
|
||||
def _classify_attribute(
|
||||
relative: str,
|
||||
node: ast.Attribute,
|
||||
parents: dict[ast.AST, ast.AST],
|
||||
bindings: _Bindings,
|
||||
legacy: set[str],
|
||||
writers: set[str],
|
||||
) -> None:
|
||||
base = _dotted(node.value)
|
||||
if base in bindings.core_modules and node.attr in FROZEN_TOP_LEVEL_V1:
|
||||
legacy.add(_site(relative, node, parents, "top_level_v1_attribute", node.attr))
|
||||
compatibility_writer_bases = {f"{module}.writer" for module in bindings.compatibility_modules}
|
||||
if base in bindings.writer_modules or base in compatibility_writer_bases:
|
||||
writers.add(_site(relative, node, parents, "writer_facade_reference", node.attr))
|
||||
|
||||
|
||||
def _classify_call(
|
||||
relative: str,
|
||||
node: ast.Call,
|
||||
parents: dict[ast.AST, ast.AST],
|
||||
bindings: _Bindings,
|
||||
writers: set[str],
|
||||
dynamic: set[str],
|
||||
) -> None:
|
||||
function = _dotted(node.func)
|
||||
if isinstance(node.func, ast.Name) and node.func.id in bindings.writer_members:
|
||||
writers.add(_site(relative, node, parents, "writer_facade_call", node.func.id))
|
||||
if isinstance(node.func, ast.Attribute):
|
||||
base = _dotted(node.func.value)
|
||||
compatibility_bases = {f"{module}.writer" for module in bindings.compatibility_modules}
|
||||
if base in bindings.writer_modules or base in compatibility_bases:
|
||||
writers.add(_site(relative, node, parents, "writer_facade_call", node.func.attr))
|
||||
if (
|
||||
_is_import_call(function, bindings)
|
||||
and (target := _import_target(node, function)) is not None
|
||||
and (target == "openultrasast.core" or target.startswith("openultrasast.core."))
|
||||
):
|
||||
dynamic.add(_site(relative, node, parents, "dynamic_core_import", target))
|
||||
if (
|
||||
function == "getattr"
|
||||
and len(node.args) >= 2
|
||||
and (name := _constant_argument(node, 1, "name")) in FROZEN_TOP_LEVEL_V1
|
||||
and _is_direct_core_object(node.args[0], bindings)
|
||||
):
|
||||
dynamic.add(_site(relative, node, parents, "dynamic_compatibility_getattr", name or ""))
|
||||
|
||||
|
||||
def _is_import_call(function: str, bindings: _Bindings) -> bool:
|
||||
if function in bindings.import_functions:
|
||||
return True
|
||||
return any(function == f"{module}.import_module" for module in bindings.import_modules)
|
||||
|
||||
|
||||
def _constant_argument(node: ast.Call, index: int, keyword: str) -> str | None:
|
||||
value: ast.AST | None = node.args[index] if len(node.args) > index else None
|
||||
if value is None:
|
||||
value = next((item.value for item in node.keywords if item.arg == keyword), None)
|
||||
return value.value if isinstance(value, ast.Constant) and isinstance(value.value, str) else None
|
||||
|
||||
|
||||
def _import_target(node: ast.Call, function: str) -> str | None:
|
||||
name = _constant_argument(node, 0, "name")
|
||||
if name is None or not name.startswith("."):
|
||||
return name
|
||||
if function == "__import__" or (package := _constant_argument(node, 1, "package")) is None:
|
||||
return None
|
||||
level = len(name) - len(name.lstrip("."))
|
||||
package_parts = package.split(".")
|
||||
if level > len(package_parts):
|
||||
return None
|
||||
resolved = [*package_parts[: len(package_parts) - level + 1]]
|
||||
if suffix := name[level:]:
|
||||
resolved.append(suffix)
|
||||
return ".".join(resolved)
|
||||
|
||||
|
||||
def _is_direct_core_object(node: ast.AST, bindings: _Bindings) -> bool:
|
||||
dotted = _dotted(node)
|
||||
if dotted in bindings.core_modules or dotted in bindings.compatibility_modules:
|
||||
return True
|
||||
if not isinstance(node, ast.Call) or not _is_import_call(_dotted(node.func), bindings):
|
||||
return False
|
||||
target = _import_target(node, _dotted(node.func))
|
||||
return target is not None and (target == "openultrasast.core" or target.startswith("openultrasast.core."))
|
||||
|
||||
|
||||
def _inspect_file(path: Path, *, root: Path) -> tuple[int, set[str], set[str], set[str]]:
|
||||
relative = relative_path(path, root)
|
||||
try:
|
||||
content = path.read_bytes()
|
||||
except OSError as exc:
|
||||
raise LegacyBoundaryError(f"boundary input is not readable: {relative}") from exc
|
||||
if not content:
|
||||
raise LegacyBoundaryError(f"boundary input must contain nonzero bytes: {relative}")
|
||||
try:
|
||||
tree = ast.parse(content, filename=relative)
|
||||
except (SyntaxError, UnicodeDecodeError) as exc:
|
||||
raise LegacyBoundaryError(f"boundary input is not parseable: {relative}") from exc
|
||||
parents = _parents(tree)
|
||||
imports = _import_sites(relative, tree, parents)
|
||||
references = _reference_sites(relative, tree, parents, _bindings(tree))
|
||||
return (
|
||||
len(content),
|
||||
imports[0] | references[0],
|
||||
imports[1] | references[1],
|
||||
imports[2] | references[2],
|
||||
)
|
||||
|
||||
|
||||
def inspect_legacy_boundary(
|
||||
paths: Sequence[Path],
|
||||
*,
|
||||
frozen_inventory: LegacyMigrationInventory,
|
||||
candidate_inventory: LegacyMigrationInventory | None = None,
|
||||
expected_scope: Sequence[Path] | None = None,
|
||||
root: Path = REPO,
|
||||
phase: BoundaryPhase = BoundaryPhase.MIGRATION,
|
||||
) -> LegacyBoundaryReport:
|
||||
"""Inspect explicit migration syntax after proving every declared input readable."""
|
||||
candidate = candidate_inventory or frozen_inventory
|
||||
validate_inventory_transition(frozen_inventory, candidate)
|
||||
legacy, writers, dynamic = set(), set(), set()
|
||||
bytes_read = 0
|
||||
unique_paths = tuple(sorted(set(paths), key=lambda value: value.as_posix()))
|
||||
if not unique_paths:
|
||||
raise LegacyBoundaryError("boundary scope must be nonempty")
|
||||
if not isinstance(phase, BoundaryPhase):
|
||||
raise LegacyBoundaryError("boundary phase must use BoundaryPhase")
|
||||
complete_paths = expected_boundary_scope(unique_paths, expected_scope, root, post_deletion=phase is BoundaryPhase.POST_DELETION)
|
||||
if complete_paths is None:
|
||||
raise LegacyBoundaryError("isolated post-deletion scope requires an explicit expected scope")
|
||||
scanned_scope, scanned_digest = scope_identity(unique_paths, root=root)
|
||||
expected_paths, expected_digest = scope_identity(complete_paths, root=root)
|
||||
for path in unique_paths:
|
||||
size, file_legacy, file_writers, file_dynamic = _inspect_file(path, root=root)
|
||||
bytes_read += size
|
||||
legacy.update(file_legacy)
|
||||
writers.update(file_writers)
|
||||
dynamic.update(file_dynamic)
|
||||
compatibility, consumer_imports, consumer_writers = partition_boundary_sites(
|
||||
legacy, writers, COMPATIBILITY_PATHS, post_deletion=phase is BoundaryPhase.POST_DELETION
|
||||
)
|
||||
open_entries = tuple(entry for entry in candidate.entries if entry.disposition is MigrationDisposition.OPEN)
|
||||
return LegacyBoundaryReport(
|
||||
phase,
|
||||
len(unique_paths),
|
||||
bytes_read,
|
||||
scanned_scope,
|
||||
scanned_digest,
|
||||
expected_paths,
|
||||
expected_digest,
|
||||
open_entries,
|
||||
tuple(sorted(legacy)),
|
||||
tuple(sorted(writers)),
|
||||
tuple(sorted(dynamic)),
|
||||
consumer_imports,
|
||||
consumer_writers,
|
||||
compatibility,
|
||||
)
|
||||
|
||||
|
||||
def require_cutover_ready(
|
||||
report: LegacyBoundaryReport,
|
||||
*,
|
||||
frozen_inventory: LegacyMigrationInventory,
|
||||
candidate_inventory: LegacyMigrationInventory,
|
||||
root: Path = REPO,
|
||||
isolated_scope: Sequence[Path] | None = None,
|
||||
) -> None:
|
||||
"""Remeasure an independently authoritative scope before allowing cutover."""
|
||||
if report.phase is not BoundaryPhase.POST_DELETION:
|
||||
raise LegacyBoundaryError("cutover requires a post-deletion boundary report")
|
||||
try:
|
||||
authoritative_scope = authoritative_cutover_scope(root, isolated_scope)
|
||||
except MigrationScopeError as exc:
|
||||
raise LegacyBoundaryError(str(exc)) from exc
|
||||
scope_error = cutover_scope_error(
|
||||
report.files_read,
|
||||
report.bytes_read,
|
||||
report.scanned_scope_paths,
|
||||
report.scanned_scope_digest,
|
||||
report.expected_scope_paths,
|
||||
report.expected_scope_digest,
|
||||
)
|
||||
if scope_error:
|
||||
raise LegacyBoundaryError(f"cutover {scope_error}")
|
||||
residue_error = cutover_residue_error(
|
||||
len(report.open_inventory_entries),
|
||||
report.compatibility_implementation_sites,
|
||||
report.legacy_import_sites or report.consumer_legacy_import_sites,
|
||||
report.legacy_writer_sites or report.consumer_legacy_writer_sites,
|
||||
report.forbidden_dynamic_sites,
|
||||
)
|
||||
if residue_error:
|
||||
raise LegacyBoundaryError(f"cutover {residue_error}")
|
||||
authoritative = inspect_legacy_boundary(
|
||||
authoritative_scope,
|
||||
frozen_inventory=frozen_inventory,
|
||||
candidate_inventory=candidate_inventory,
|
||||
expected_scope=authoritative_scope,
|
||||
root=root,
|
||||
phase=BoundaryPhase.POST_DELETION,
|
||||
)
|
||||
if report != authoritative:
|
||||
raise LegacyBoundaryError("cutover report does not match the authoritative boundary scan")
|
||||
|
||||
|
||||
__all__ = [
|
||||
"BoundaryPhase",
|
||||
"FROZEN_TOP_LEVEL_V1",
|
||||
"LegacyBoundaryError",
|
||||
"LegacyBoundaryReport",
|
||||
"boundary_python",
|
||||
"inspect_legacy_boundary",
|
||||
"require_cutover_ready",
|
||||
"static_top_level_v1",
|
||||
]
|
||||
Reference in New Issue
Block a user