Zeroize Audit

tools/scripts/semantic_audit.py

#!/usr/bin/env python3
# /// script
# requires-python = ">=3.11"
# dependencies = []
# ///
"""
semantic_audit.py — Rust trait-aware zeroization auditor.

Reads rustdoc JSON (generated by `cargo +nightly rustdoc --document-private-items
-- -Z unstable-options --output-format json`) and emits findings about missing or
incorrect zeroization of sensitive types.

Usage:
    uv run semantic_audit.py --rustdoc <path.json> [--cargo-toml <Cargo.toml>] --out <findings.json>

Exit codes:
    0  — ran successfully (findings may be empty)
    1  — rustdoc JSON not found or unparseable
    2  — argument error
"""

import argparse
import json
import re
import sys
import tomllib
from pathlib import Path
from typing import Any

# ---------------------------------------------------------------------------
# Sensitive type / field name patterns
# ---------------------------------------------------------------------------

SENSITIVE_TYPE_RE = re.compile(
    r"(?i)(Key|PrivateKey|SecretKey|SigningKey|MasterKey|HmacKey|"
    r"Password|Passphrase|Pin|Token|AuthToken|BearerToken|ApiKey|"
    r"Secret|SharedSecret|PreSharedKey|Nonce|Seed|Entropy|"
    r"Credential|SessionKey|DerivedKey)"
)

SENSITIVE_FIELD_RE = re.compile(
    r"(?i)\b(key|secret|password|token|nonce|seed|private|master|credential)\b"
)

# Derives/traits that indicate zeroization intent
ZEROIZE_TRAITS = {"Zeroize", "ZeroizeOnDrop"}
DROP_TRAIT = "Drop"

# Traits / derives that create untracked copies
COPY_DERIVES = {"Copy"}
CLONE_DERIVES = {"Clone"}
DEBUG_DERIVES = {"Debug"}
SERIALIZE_DERIVES = {"Serialize"}

# Evidence tags used for conservative confidence mapping.
STRONG_EVIDENCE_TAGS = {
    "trait_impl",
    "resolved_path",
    "drop_body_source",
    "cargo_toml",
}

MEDIUM_EVIDENCE_TAGS = {
    "source_scan",
    "generic_traversal",
}

HEAP_TYPE_NAMES = {
    "Vec",
    "Box",
    "String",
    "HashMap",
    "BTreeMap",
    "VecDeque",
    "BinaryHeap",
    "LinkedList",
}

ZEROIZING_WRAPPER_NAMES = {
    "Zeroizing",
}

MANUALLY_DROP_NAMES = {"ManuallyDrop"}

ZEROIZING_NAME_HINT_RE = re.compile(r"(?i)(Zeroiz|Protected|Secret|Sensitive)")


# ---------------------------------------------------------------------------
# Helper: is a type name sensitive?
# ---------------------------------------------------------------------------


def is_sensitive_name(name: str) -> bool:
    return bool(SENSITIVE_TYPE_RE.search(name))


def has_sensitive_field(fields: list[dict]) -> bool:
    for field in fields:
        fname = field.get("name") or ""
        if SENSITIVE_FIELD_RE.search(fname):
            return True
    return False


# ---------------------------------------------------------------------------
# Finding builder
# ---------------------------------------------------------------------------

_finding_counter = [0]


def make_finding(
    category: str,
    severity: str,
    detail: str,
    type_name: str,
    file: str,
    line: int | None,
    confidence: str | None = None,
    evidence_strength: list[str] | None = None,
) -> dict:
    _finding_counter[0] += 1
    fid = f"F-RUST-SRC-{_finding_counter[0]:04d}"
    evidence_strength = evidence_strength or ["heuristic"]
    resolved_confidence = confidence or _confidence_from_evidence_strength(evidence_strength)
    return {
        "id": fid,
        "language": "rust",
        "category": category,
        "severity": severity,
        "confidence": resolved_confidence,
        "evidence_strength": evidence_strength,
        "detail": detail,
        "symbol": type_name,
        "object": {"name": type_name},
        "location": {"file": file, "line": line or 1},
        "evidence": [
            {
                "source": "rustdoc_json",
                "detail": detail,
                "strength": evidence_strength,
            }
        ],
    }


def _confidence_from_evidence_strength(evidence_strength: list[str]) -> str:
    strong_count = sum(1 for tag in evidence_strength if tag in STRONG_EVIDENCE_TAGS)
    medium_count = sum(1 for tag in evidence_strength if tag in MEDIUM_EVIDENCE_TAGS)
    if strong_count >= 2:
        return "confirmed"
    if strong_count == 1:
        return "likely"
    if medium_count >= 1 and not any(tag == "heuristic" for tag in evidence_strength):
        return "likely"
    return "needs_review"


# ---------------------------------------------------------------------------
# Rustdoc JSON helpers
# ---------------------------------------------------------------------------


def item_span(item: dict) -> tuple[str, int | None]:
    """Return (file, line) from an item's span."""
    span = item.get("span") or {}
    filename = span.get("filename") or ""
    begin = span.get("begin") or []
    line = begin[0] if begin else None
    return filename, line


def item_derives(item: dict) -> set[str]:
    """Collect derive macro names from item attrs."""
    derives: set[str] = set()
    for attr in item.get("attrs") or []:
        # attr is a string like '#[derive(Copy, Clone, Debug)]'
        m = re.search(r"derive\(([^)]+)\)", attr)
        if m:
            for d in m.group(1).split(","):
                derives.add(d.strip())
    return derives


def item_impls(item: dict, index: dict) -> set[str]:
    """Return trait names implemented by this struct/enum via its impl IDs."""
    trait_names: set[str] = set()
    for impl_id in item.get("impls") or []:
        impl_item = index.get(str(impl_id)) or {}
        inner = impl_item.get("inner") or {}
        impl_data = inner.get("impl") or {}
        trait_ref = impl_data.get("trait") or {}
        tname = _trait_name(trait_ref)
        if tname:
            trait_names.add(tname)
    return trait_names


def _trait_name(trait_ref: dict[str, Any]) -> str:
    name = trait_ref.get("name")
    if isinstance(name, str) and name:
        return name.split("::")[-1]
    resolved = trait_ref.get("resolved_path")
    if isinstance(resolved, dict):
        resolved_name = resolved.get("name")
        if isinstance(resolved_name, str) and resolved_name:
            return resolved_name.split("::")[-1]
    return ""


def struct_fields(item: dict, index: dict) -> list[dict]:
    """Return field items for a struct."""
    fields: list[dict] = []
    inner = item.get("inner") or {}
    struct_data = inner.get("struct") or {}
    kind = struct_data.get("kind") or {}
    # plain struct: kind = {"plain": {"fields": [id, ...], ...}}
    plain = kind.get("plain") or {}
    field_ids = plain.get("fields") or []
    for fid in field_ids:
        fitem = index.get(str(fid)) or {}
        fields.append(fitem)
    return fields


# ---------------------------------------------------------------------------
# Core analysis
# ---------------------------------------------------------------------------


def analyze(rustdoc: dict, cargo_toml_path: str | None) -> list[dict]:
    findings: list[dict] = []
    index: dict = rustdoc.get("index") or {}

    # Check whether zeroize crate is a dependency
    has_zeroize_dep = _check_zeroize_dep(cargo_toml_path)

    for _item_id, item in index.items():
        kind = item.get("kind") or ""
        if kind not in ("struct", "enum"):
            continue

        name = item.get("name") or ""
        if not is_sensitive_name(name):
            # Check fields too
            fields = struct_fields(item, index) if kind == "struct" else []
            if not has_sensitive_field(fields):
                continue

        file, line = item_span(item)
        derives = item_derives(item)
        trait_impls = item_impls(item, index)

        # --- 1. Copy derive on sensitive type ---
        if COPY_DERIVES & derives:
            findings.append(
                make_finding(
                    "SECRET_COPY",
                    "critical",
                    f"#[derive(Copy)] on sensitive type '{name}' — all assignments are "
                    "untracked duplicates, no Drop ever runs",
                    name,
                    file,
                    line,
                    evidence_strength=["attr_only", "sensitive_name_or_field"],
                )
            )

        # --- 2. No Zeroize / ZeroizeOnDrop / Drop ---
        # (Skip for Copy types: Copy and Drop are mutually exclusive in Rust.)
        has_zeroize = bool(ZEROIZE_TRAITS & trait_impls)
        has_drop = DROP_TRAIT in trait_impls
        has_zeroize_on_drop = "ZeroizeOnDrop" in trait_impls or "ZeroizeOnDrop" in derives

        if not (COPY_DERIVES & derives):
            if not has_zeroize and not has_drop and not has_zeroize_on_drop:
                findings.append(
                    make_finding(
                        "MISSING_SOURCE_ZEROIZE",
                        "high",
                        f"Sensitive type '{name}' has no Zeroize, ZeroizeOnDrop,"
                        " or Drop implementation",
                        name,
                        file,
                        line,
                        evidence_strength=["trait_impl", "sensitive_name_or_field"],
                    )
                )
            elif has_zeroize and not has_zeroize_on_drop and not has_drop:
                # Zeroize implemented but never auto-triggered
                findings.append(
                    make_finding(
                        "MISSING_SOURCE_ZEROIZE",
                        "high",
                        f"Sensitive type '{name}' implements Zeroize but has no "
                        "ZeroizeOnDrop or Drop to trigger it automatically",
                        name,
                        file,
                        line,
                        evidence_strength=["trait_impl", "sensitive_name_or_field"],
                    )
                )

        # --- 3. Partial Drop: Drop impl present but not all secret fields zeroed ---
        if has_drop and kind == "struct":
            fields = struct_fields(item, index)
            secret_fields = [f for f in fields if SENSITIVE_FIELD_RE.search(f.get("name") or "")]
            if secret_fields:
                # Find the Drop impl and check whether it zeroes all secret fields.
                drop_impls = _find_drop_impl_items(item, index)
                if drop_impls:
                    secret_field_names = [f.get("name") or "" for f in secret_fields]
                    zeroed_names, evidence_strength = _zeroed_field_names_in_drop(
                        drop_impls[0], index, secret_field_names
                    )
                    unzeroed = [
                        f.get("name") for f in secret_fields if f.get("name") not in zeroed_names
                    ]
                    if unzeroed:
                        severity = "high" if "drop_body_source" in evidence_strength else "medium"
                        findings.append(
                            make_finding(
                                "PARTIAL_WIPE",
                                severity,
                                f"Drop impl for '{name}' does not zero all secret fields: "
                                f"missing {unzeroed}",
                                name,
                                file,
                                line,
                                evidence_strength=evidence_strength + ["trait_impl"],
                            )
                        )
                    elif "drop_body_source" not in evidence_strength:
                        findings.append(
                            make_finding(
                                "PARTIAL_WIPE",
                                "medium",
                                f"Drop impl for '{name}' found, but field-level "
                                "zeroization could not be confirmed from "
                                "function body; review manually",
                                name,
                                file,
                                line,
                                evidence_strength=evidence_strength + ["trait_impl"],
                            )
                        )

        # --- 4. ZeroizeOnDrop with heap (Vec/Box) fields ---
        if has_zeroize_on_drop and kind == "struct":
            fields = struct_fields(item, index)
            heap_fields = _heap_fields(fields, index, source_file=file)
            alias_review = "__alias_review__" in heap_fields
            real_heap_fields = [f for f in heap_fields if f != "__alias_review__"]
            if real_heap_fields:
                findings.append(
                    make_finding(
                        "PARTIAL_WIPE",
                        "medium",
                        f"ZeroizeOnDrop on '{name}' which has heap fields {real_heap_fields} — "
                        "capacity bytes beyond len may not be zeroed",
                        name,
                        file,
                        line,
                        evidence_strength=["resolved_path", "generic_traversal", "trait_impl"],
                    )
                )
            elif alias_review:
                findings.append(
                    make_finding(
                        "PARTIAL_WIPE",
                        "medium",
                        f"ZeroizeOnDrop on '{name}' — source file contains type aliases that may "
                        "wrap heap types (Vec/Box/String); verify all heap fields are covered",
                        name,
                        file,
                        line,
                        evidence_strength=["alias_heuristic", "source_scan", "trait_impl"],
                    )
                )

        # --- 4b. ManuallyDrop<T> field on sensitive struct ---
        if kind == "struct":
            fields = struct_fields(item, index)
            md_fields = _manually_drop_fields(fields, index)
            if md_fields:
                findings.append(
                    make_finding(
                        "MISSING_SOURCE_ZEROIZE",
                        "critical",
                        f"Sensitive struct '{name}' has ManuallyDrop<T> field(s) {md_fields} — "
                        "Drop does not run automatically on ManuallyDrop fields; "
                        "secret is not zeroed unless ManuallyDrop::drop() is called explicitly",
                        name,
                        file,
                        line,
                        evidence_strength=["resolved_path", "trait_impl"],
                    )
                )

        # --- 5. Clone on zeroizing type ---
        if CLONE_DERIVES & derives and (has_zeroize or has_zeroize_on_drop or has_drop):
            findings.append(
                make_finding(
                    "SECRET_COPY",
                    "medium",
                    f"Clone on zeroizing type '{name}' — each clone is an independent allocation "
                    "that must be independently zeroed",
                    name,
                    file,
                    line,
                    evidence_strength=["attr_only", "trait_impl"],
                )
            )

        # --- 6. From/Into returning non-zeroizing type ---
        from_into_escapes = _find_from_into_non_zeroizing(item, index)
        for escape, evidence_strength in from_into_escapes:
            findings.append(
                make_finding(
                    "SECRET_COPY",
                    "medium",
                    f"'{name}' has {escape} conversion returning a non-zeroizing type — "
                    "bytes escape into caller's ownership in a non-zeroizing container",
                    name,
                    file,
                    line,
                    evidence_strength=evidence_strength + ["trait_impl"],
                )
            )

        # --- 7. ptr::write_bytes without compiler_fence ---
        if _has_write_bytes_without_compiler_fence(file):
            findings.append(
                make_finding(
                    "OPTIMIZED_AWAY_ZEROIZE",
                    "medium",
                    f"'{name}' is defined in a file that uses ptr::write_bytes without "
                    "compiler_fence — wipe may be optimized away by the compiler",
                    name,
                    file,
                    line,
                    evidence_strength=["source_scan", "heuristic"],
                )
            )

        # --- 8. cfg(feature) wrapping Drop/Zeroize ---
        if _has_cfg_feature_on_cleanup(item, index):
            findings.append(
                make_finding(
                    "NOT_ON_ALL_PATHS",
                    "medium",
                    f"#[cfg(feature=...)] wraps Drop or Zeroize impl for '{name}' — "
                    "zeroing absent when feature flag is off",
                    name,
                    file,
                    line,
                    evidence_strength=["attr_only", "trait_impl"],
                )
            )

        # --- 9. Debug derive ---
        if DEBUG_DERIVES & derives:
            findings.append(
                make_finding(
                    "SECRET_COPY",
                    "low",
                    f"#[derive(Debug)] on sensitive type '{name}' — "
                    "secrets may appear in formatted output / log entries",
                    name,
                    file,
                    line,
                    evidence_strength=["attr_only"],
                )
            )

        # --- 10. Serialize derive ---
        if SERIALIZE_DERIVES & derives:
            findings.append(
                make_finding(
                    "SECRET_COPY",
                    "low",
                    f"#[derive(Serialize)] on sensitive type '{name}' — "
                    "serialization creates an uncontrolled copy of secret bytes",
                    name,
                    file,
                    line,
                    evidence_strength=["attr_only"],
                )
            )

    # --- 11. No zeroize crate dependency ---
    # Only emit when Cargo.toml was provided and successfully parsed but did
    # not list zeroize.  has_zeroize_dep is None when the path was omitted or
    # the file could not be parsed, which must not trigger a false finding.
    if has_zeroize_dep is False:
        findings.append(
            make_finding(
                "MISSING_SOURCE_ZEROIZE",
                "low",
                "No 'zeroize' crate in Cargo.toml dependencies — "
                "all manual zeroing lacks approved-API guarantee",
                "<crate>",
                str(cargo_toml_path or "Cargo.toml"),
                1,
                evidence_strength=["cargo_toml"],
            )
        )

    return findings


# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------


def _check_zeroize_dep(cargo_toml_path: str | None) -> bool | None:
    """Return True/False if Cargo.toml was parsed, None if path absent or unreadable."""
    if not cargo_toml_path:
        return None
    try:
        content = Path(cargo_toml_path).read_text(encoding="utf-8")
        manifest = tomllib.loads(content)
    except OSError:
        return None
    except tomllib.TOMLDecodeError as e:
        print(
            f"semantic_audit.py: warning: cannot parse Cargo.toml {cargo_toml_path!r}: {e}",
            file=sys.stderr,
        )
        return None
    return _manifest_has_zeroize_dep(manifest)


def _manifest_has_zeroize_dep(manifest: dict) -> bool:
    return any(_dep_table_has_zeroize(dep_table) for dep_table in _iter_dependency_tables(manifest))


def _iter_dependency_tables(manifest: dict) -> list[dict]:
    dep_tables: list[dict] = []

    dependencies = manifest.get("dependencies")
    if isinstance(dependencies, dict):
        dep_tables.append(dependencies)

    workspace = manifest.get("workspace")
    if isinstance(workspace, dict):
        workspace_deps = workspace.get("dependencies")
        if isinstance(workspace_deps, dict):
            dep_tables.append(workspace_deps)

    target = manifest.get("target")
    if isinstance(target, dict):
        for target_data in target.values():
            if not isinstance(target_data, dict):
                continue
            target_deps = target_data.get("dependencies")
            if isinstance(target_deps, dict):
                dep_tables.append(target_deps)

    return dep_tables


def _dep_table_has_zeroize(dep_table: dict) -> bool:
    for dep_name, dep_spec in dep_table.items():
        if isinstance(dep_name, str) and dep_name.lower() == "zeroize":
            return True
        if isinstance(dep_spec, dict):
            package_name = dep_spec.get("package")
            if isinstance(package_name, str) and package_name.lower() == "zeroize":
                return True
    return False


def _find_drop_impl_items(item: dict, index: dict) -> list[dict]:
    result = []
    for impl_id in item.get("impls") or []:
        impl_item = index.get(str(impl_id)) or {}
        inner = impl_item.get("inner") or {}
        impl_data = inner.get("impl") or {}
        trait_ref = impl_data.get("trait") or {}
        if _trait_name(trait_ref) == "Drop":
            result.append(impl_item)
    return result


def _zeroed_field_names_in_drop(
    drop_impl: dict, index: dict, secret_fields: list[str]
) -> tuple[set[str], list[str]]:
    """
    Extract zeroed fields from Drop evidence.

    Prefers parsing Drop::drop source span. Falls back to docs text when source
    body is unavailable.
    """
    body = _extract_drop_body_from_impl(drop_impl, index)
    if body:
        return _zeroed_field_names_in_text(body, secret_fields), ["drop_body_source"]

    docs = drop_impl.get("docs") or ""
    if docs:
        return _zeroed_field_names_in_text(docs, secret_fields), ["docs_heuristic"]

    return set(), ["unavailable"]


def _extract_drop_body_from_impl(drop_impl: dict, index: dict) -> str:
    inner = drop_impl.get("inner") or {}
    impl_data = inner.get("impl") or {}
    for method_id in impl_data.get("items") or []:
        method_item = index.get(str(method_id)) or {}
        if (method_item.get("kind") or "") != "function":
            continue
        if (method_item.get("name") or "") != "drop":
            continue
        source = _read_item_span_source(method_item)
        if source:
            return source
    return ""


def _read_item_span_source(item: dict) -> str:
    span = item.get("span") or {}
    filename = span.get("filename")
    begin = span.get("begin") or []
    end = span.get("end") or []
    if not filename or not begin or not end:
        return ""
    try:
        lines = Path(filename).read_text(encoding="utf-8", errors="replace").splitlines()
    except OSError as e:
        print(
            f"semantic_audit.py: warning: cannot read span source {filename!r}: {e}",
            file=sys.stderr,
        )
        return ""

    start_line = max(int(begin[0]), 1)
    end_line = max(int(end[0]), start_line)
    if start_line > len(lines):
        return ""
    snippet = lines[start_line - 1 : min(end_line, len(lines))]
    return "\n".join(snippet)


def _zeroed_field_names_in_text(text: str, field_names: list[str]) -> set[str]:
    zeroed: set[str] = set()
    for field_name in field_names:
        escaped = re.escape(field_name)
        patterns = [
            rf"\bself\.{escaped}\.zeroize\s*\(",
            rf"\bzeroize\s*\(\s*&mut\s+self\.{escaped}\s*\)",
            rf"\bself\.{escaped}\s*=\s*(?:0+|Default::default\(\)|\[[^]]+\])",
            rf"\bself\.{escaped}\.fill\s*\(\s*0\s*\)",
        ]
        if any(re.search(pattern, text) for pattern in patterns):
            zeroed.add(field_name)
    return zeroed


# Matches type alias definitions like: type SecretBuffer = Vec<u8>;
_TYPE_ALIAS_RE = re.compile(
    r"^\s*(?:pub\s+)?type\s+\w+\s*=\s*(?:Vec|Box|String|HashMap|BTreeMap)\b"
)


def _heap_fields(fields: list[dict], index: dict, source_file: str | None = None) -> list[str]:
    heap: list[str] = []
    for field in fields:
        fname = field.get("name") or ""
        inner = field.get("inner") or {}
        struct_field = inner.get("struct_field") or {}
        ty = struct_field.get("type") or {}
        if _type_contains_heap(ty, index):
            heap.append(fname)
    # If no heap fields found via rustdoc, scan the source file for type aliases
    # that may wrap heap types (e.g. `type SecretBuffer = Vec<u8>`). Emit a
    # needs_review note by appending a sentinel value so callers can detect this.
    if not heap and source_file:
        try:
            src = Path(source_file).read_text(encoding="utf-8", errors="replace")
            if _TYPE_ALIAS_RE.search(src):
                heap.append("__alias_review__")
        except OSError as e:
            print(
                f"semantic_audit.py: warning: cannot read source"
                f" for alias scan {source_file!r}: {e}",
                file=sys.stderr,
            )
    return heap


def _manually_drop_fields(fields: list[dict], index: dict) -> list[str]:
    """Return field names whose type is or contains ManuallyDrop<T>."""
    result: list[str] = []
    for field in fields:
        fname = field.get("name") or ""
        inner = field.get("inner") or {}
        struct_field = inner.get("struct_field") or {}
        ty = struct_field.get("type") or {}
        names = _type_named_paths(ty, index, set())
        if MANUALLY_DROP_NAMES & names:
            result.append(fname)
    return result


def _find_from_into_non_zeroizing(item: dict, index: dict) -> list[tuple[str, list[str]]]:
    escapes: list[tuple[str, list[str]]] = []
    for impl_id in item.get("impls") or []:
        impl_item = index.get(str(impl_id)) or {}
        inner = impl_item.get("inner") or {}
        impl_data = inner.get("impl") or {}
        trait_ref = impl_data.get("trait") or {}
        tname = _trait_name(trait_ref)
        if tname not in ("From", "Into"):
            continue
        for target_type in _iter_trait_type_args(trait_ref):
            if _type_is_zeroizing(target_type, index):
                continue
            target_desc = _type_description(target_type, index)
            evidence = (
                ["resolved_path", "generic_traversal"]
                if _type_has_resolved_path(target_type)
                else ["alias_heuristic"]
            )
            escapes.append((f"{tname}<{target_desc}>", evidence))
    return escapes


def _iter_trait_type_args(trait_ref: dict) -> list[dict]:
    args = trait_ref.get("args") or {}
    angle = args.get("angle_bracketed") or {}
    out: list[dict] = []
    for arg in angle.get("args") or []:
        ty = arg.get("type")
        if isinstance(ty, dict):
            out.append(ty)
    return out


def _type_contains_heap(ty: dict[str, Any], index: dict, seen: set[str] | None = None) -> bool:
    seen = seen or set()
    return any(name in HEAP_TYPE_NAMES for name in _type_named_paths(ty, index, seen))


def _type_is_zeroizing(ty: dict[str, Any], index: dict, seen: set[str] | None = None) -> bool:
    seen = seen or set()
    names = _type_named_paths(ty, index, seen)
    if any(name in ZEROIZING_WRAPPER_NAMES for name in names):
        return True
    return any(ZEROIZING_NAME_HINT_RE.search(name) for name in names)


def _type_has_resolved_path(ty: dict[str, Any]) -> bool:
    if not isinstance(ty, dict):
        return False
    if "resolved_path" in ty:
        return True
    return any(_type_has_resolved_path(nested) for nested in _iter_nested_types(ty))


def _type_description(ty: dict[str, Any], index: dict) -> str:
    names = sorted(_type_named_paths(ty, index, set()))
    if names:
        return "::".join(names[:2]) if len(names) > 1 else names[0]
    return "unknown"


def _type_named_paths(ty: dict[str, Any], index: dict, seen_alias_ids: set[str]) -> set[str]:
    names: set[str] = set()
    if not isinstance(ty, dict):
        return names

    resolved = ty.get("resolved_path")
    if isinstance(resolved, dict):
        raw_name = resolved.get("name")
        if isinstance(raw_name, str) and raw_name:
            names.add(raw_name.split("::")[-1])

        alias_id = resolved.get("id")
        alias_item = index.get(str(alias_id)) if alias_id is not None else None
        alias_id_str = str(alias_id) if alias_id is not None else ""
        if (
            alias_id_str
            and alias_id_str not in seen_alias_ids
            and isinstance(alias_item, dict)
            and (alias_item.get("kind") or "") == "typedef"
        ):
            seen_alias_ids.add(alias_id_str)
            alias_type = ((alias_item.get("inner") or {}).get("type_alias") or {}).get("type") or {}
            names |= _type_named_paths(alias_type, index, seen_alias_ids)

        args = resolved.get("args") or {}
        names |= _type_args_named_paths(args, index, seen_alias_ids)

    for nested in _iter_nested_types(ty):
        names |= _type_named_paths(nested, index, seen_alias_ids)
    return names


def _type_args_named_paths(args: dict[str, Any], index: dict, seen_alias_ids: set[str]) -> set[str]:
    names: set[str] = set()
    angle = args.get("angle_bracketed") if isinstance(args, dict) else None
    if not isinstance(angle, dict):
        return names
    for arg in angle.get("args") or []:
        if isinstance(arg, dict):
            ty = arg.get("type")
            if isinstance(ty, dict):
                names |= _type_named_paths(ty, index, seen_alias_ids)
    return names


def _iter_nested_types(ty: dict[str, Any]) -> list[dict[str, Any]]:
    nested: list[dict[str, Any]] = []

    borrowed = ty.get("borrowed_ref")
    if isinstance(borrowed, dict):
        inner_ty = borrowed.get("type")
        if isinstance(inner_ty, dict):
            nested.append(inner_ty)

    raw_ptr = ty.get("raw_pointer")
    if isinstance(raw_ptr, dict):
        inner_ty = raw_ptr.get("type")
        if isinstance(inner_ty, dict):
            nested.append(inner_ty)

    array_ty = ty.get("array")
    if isinstance(array_ty, dict):
        inner_ty = array_ty.get("type")
        if isinstance(inner_ty, dict):
            nested.append(inner_ty)

    slice_ty = ty.get("slice")
    if isinstance(slice_ty, dict):
        nested.append(slice_ty)

    tuple_types = ty.get("tuple")
    if isinstance(tuple_types, list):
        for inner_ty in tuple_types:
            if isinstance(inner_ty, dict):
                nested.append(inner_ty)

    qualified = ty.get("qualified_path")
    if isinstance(qualified, dict):
        qself = qualified.get("self_type")
        if isinstance(qself, dict):
            nested.append(qself)
        qtrait = qualified.get("trait")
        if isinstance(qtrait, dict):
            nested.append(qtrait)

    return nested


_COMPILER_FENCE_RE = re.compile(
    r"\b(?:core::sync::atomic::|std::sync::atomic::)?compiler_fence\s*\("
)


def _has_write_bytes_without_compiler_fence(source_file: str | None) -> bool:
    if not source_file:
        return False
    try:
        src = Path(source_file).read_text(encoding="utf-8", errors="replace")
    except OSError:
        return False
    return "write_bytes" in src and not _COMPILER_FENCE_RE.search(src)


def _has_cfg_feature_on_cleanup(item: dict, index: dict) -> bool:
    for impl_id in item.get("impls") or []:
        impl_item = index.get(str(impl_id)) or {}
        inner = impl_item.get("inner") or {}
        impl_data = inner.get("impl") or {}
        trait_ref = impl_data.get("trait") or {}
        tname = _trait_name(trait_ref)
        if tname not in ("Drop", "Zeroize", "ZeroizeOnDrop"):
            continue
        for attr in impl_item.get("attrs") or []:
            if "cfg" in attr and "feature" in attr:
                return True
    return False


# ---------------------------------------------------------------------------
# CLI
# ---------------------------------------------------------------------------


def main() -> int:
    parser = argparse.ArgumentParser(
        description="Rust trait-aware zeroization auditor (rustdoc JSON input)"
    )
    parser.add_argument("--rustdoc", required=True, help="Path to rustdoc JSON file")
    parser.add_argument("--cargo-toml", help="Path to Cargo.toml (for dependency checks)")
    parser.add_argument("--out", required=True, help="Output findings JSON path")
    args = parser.parse_args()

    rustdoc_path = Path(args.rustdoc)
    if not rustdoc_path.exists():
        print(f"semantic_audit.py: rustdoc JSON not found: {rustdoc_path}", file=sys.stderr)
        return 1

    try:
        rustdoc = json.loads(rustdoc_path.read_text(encoding="utf-8"))
    except (json.JSONDecodeError, OSError) as e:
        print(f"semantic_audit.py: failed to parse rustdoc JSON: {e}", file=sys.stderr)
        return 1

    findings = analyze(rustdoc, args.cargo_toml)

    out_path = Path(args.out)
    out_path.parent.mkdir(parents=True, exist_ok=True)
    out_path.write_text(json.dumps(findings, indent=2), encoding="utf-8")

    print(f"semantic_audit.py: {len(findings)} finding(s) written to {out_path}")
    return 0


if __name__ == "__main__":
    sys.exit(main())