鸽子的回忆录

def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("--panel", required=False)
    parser.add_argument("--report-only", action="store_true")
    args = parser.parse_args()
    node_actions = get_all_actions()
    missing = []
    if args.panel:
        panel_root = Path(args.panel)
        sys.path.insert(0, str(panel_root))
        from apps.node_manager.protocol import get_all_actions as panel_get_all
        panel_actions = panel_get_all()
        for action, spec in panel_actions.items():
            direction = spec.direction.value if isinstance(spec.direction, Enum) else str(spec.direction)
            if direction in {"server_to_node", "bidirectional"} and action not in node_actions:
                missing.append(action)
    print(json.dumps({
        "node_actions": len(node_actions),
        "missing_panel_to_node_actions": missing,
    }, ensure_ascii=False, indent=2))
    return 0 if args.report_only or not missing else 1

def dynamic_usage():
    findings = []
    for path in sorted(ROOT.glob("**/*.py")):
        if any(part in {".venv", "__pycache__", ".git"} for part in path.parts):
            continue
        try:
            tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path))
        except (SyntaxError, UnicodeDecodeError):
            continue
        rel = path.relative_to(ROOT).as_posix()
        for node in ast.walk(tree):
            if isinstance(node, ast.Call):
                name = dotted(node.func).rsplit(".", 1)[-1]
                if name in {"getattr", "setattr", "hasattr", "__import__"}:
                    findings.append({"path": rel, "line": node.lineno, "symbol": name})
    return findings

def dotted(node):
    if isinstance(node, ast.Name):
        return node.id
    if isinstance(node, ast.Attribute):
        parent = dotted(node.value)
        return f"{parent}.{node.attr}" if parent else node.attr
    return ""

print(json.dumps({"status": "security matrix scaffold", "cases": []}, ensure_ascii=False, indent=2))

root = Path(__file__).resolve().parents[1]
tests = sorted(path.relative_to(root).as_posix() for path in root.glob("tests/**/*.py"))
print(json.dumps({"tests": tests}, ensure_ascii=False, indent=2))

actions = [asdict(spec) for spec in get_all_actions().values()]
Path(args.output).write_text(json.dumps(actions, ensure_ascii=False, indent=2), encoding="utf-8")

async def dispatch_docker(self, action: str, payload, raw_message: dict) -> bool:
    if not await self.verify_if_required(action, payload, raw_message, label="Docker动作"):
        return False
    await self.docker_manager_action(action, payload)

async def docker_manager_action(self, action: str, data):
    parts = action.split(":")
    if len(parts) != 3 or parts[0] != "docker":
        logger.warning(f"非法Docker动作: {action}")
        return
    await self.client.services.docker_manager.handle(parts[1], parts[2], data or {})

async def handle(self, resource: str, operation: str, payload: dict):
    if not self.__safe_enabled("docker_manager", False):
        return await self.__send_error(payload, resource, operation, "Docker管理未启用")

    read_operations = {"list", "inspect", "history", "stats", "logs", "info", "df", "version", "events", "get", "validate"}
    is_read = operation in read_operations or (
        resource == "compose" and operation in {"list", "ps", "logs", "config"}
    )

    if not is_read:
        required_flag = RESOURCE_FLAGS.get(resource)
        if required_flag and not self.__safe_enabled(required_flag, False):
            return await self.__send_error(payload, resource, operation, "该Docker写操作未启用")

RESOURCE_FLAGS = {
    "image": "docker_image_manage",
    "container": "docker_container_manage",
    "network": "docker_network_manage",
    "volume": "docker_volume_manage",
    "config": "docker_config_manage",
}

result = await asyncio.to_thread(self.__execute, resource, operation, payload)
await self.__send_success(payload, resource, operation, result)

def __ensure_container_image(self, client, image: str, payload: dict):
    try:
        client.images.get(image)
        return None  # 本地有，直接用
    except (ImageNotFound, NotFound):
        pass

    if not self.__coerce_bool(payload.get("pull_missing"), True):
        raise RuntimeError(f"节点本机缺少镜像 {image}，请先拉取镜像或启用自动拉取")
    if not self.__safe_enabled("docker_image_manage", False):
        raise RuntimeError("节点本机缺少镜像，自动拉取需要启用 docker_image_manage")
    return self.__pull_image_reference(client, image, payload)

def __compose_command(self, project_dir: Path, command: list[str],
                      json_lines: bool = False, check: bool = True):
    if not shutil.which("docker"):
        raise RuntimeError("当前系统未安装docker命令")
    result = subprocess.run(
        ["docker", "compose", *command],
        cwd=str(project_dir),
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        text=True,
        timeout=300,
    )
    # ...

def __compose_project_dir(self, project: str, create: bool = False) -> Path:
    project = str(project or "").strip()
    if not project.replace("-", "").replace("_", "").isalnum():
        raise ValueError("Compose项目名只能包含字母、数字、下划线和短横线")
    base = COMPOSE_BASE_DIR.resolve()
    project_dir = (base / project).resolve()
    if base != project_dir and base not in project_dir.parents:
        raise ValueError("Compose项目路径非法")

def __container_exec(self, payload: dict):
    command = payload.get("command") or payload.get("cmd")
    if isinstance(command, str):
        command = command.strip()
    if not command:
        raise ValueError("容器执行命令不能为空")
    result = self.__get_container(payload).exec_run(
        command,
        stdout=True, stderr=True,
        tty=self.__coerce_bool(payload.get("tty"), False),
        privileged=self.__coerce_bool(payload.get("privileged"), False),
        user=payload.get("user") or "",
        workdir=payload.get("workdir") or None,
        environment=payload.get("environment") if isinstance(payload.get("environment"), dict) else None,
    )

def __config(self, operation: str, payload: dict):
    if operation == "update":
        if os.name == "nt":
            raise RuntimeError("Windows节点暂不支持修改daemon.json")
        data = self.__daemon_config_data(payload)
        DAEMON_CONFIG_PATH.parent.mkdir(parents=True, exist_ok=True)
        if DAEMON_CONFIG_PATH.exists():
            backup = DAEMON_CONFIG_PATH.with_suffix(f".json.bak")
            shutil.copy2(DAEMON_CONFIG_PATH, backup)
        DAEMON_CONFIG_PATH.write_text(
            json.dumps(data, ensure_ascii=False, indent=2) + "\n",
            encoding="utf-8"
        )

@staticmethod
def __json_safe(value: Any):
    try:
        return json.loads(json.dumps(value, default=str))
    except Exception:
        return str(value)

@staticmethod
def __required(payload: dict, *keys: str) -> Any:
    for key in keys:
        value = payload.get(key)
        if value not in (None, ""):
            return value
    raise ValueError(f"缺少参数: {'/'.join(keys)}")

async def handle_action(self, action: str, data: dict):
    handler = getattr(self, f"handle_{action}", None)
    if handler:
        await handler(data)
    else:
        logger.warning(f"Unknown action: {action}")

async def handle_execute_command(self, data):
    sign_data = data.get("_sign")
    if not sign_data:
        return await self.close_ws(3001)
    if not verify_signature(self.node_token(), "execute:run_shell", data, sign_data):
        return await self.close_ws(3001)
    # ... 然后才是业务逻辑

class NodeActionDispatcher:
    def __init__(self, client):
        self.client = client
        self.actions = MappingProxyType({
            "node:close": self.close,
            "node:init_config": self.init_node_config,
            "terminal:create_session": self.terminal_create_session,
            "terminal:close_session": self.terminal_close_session,
            "terminal:input": self.terminal_input,
            "terminal:resize": self.terminal_resize,
            "execute:run_shell": self.execute_shell,
            "file_manager:list": self.file_manager_list,
            "file_manager:read": self.file_manager_read,
            # ...
        })

async def dispatch(self, action: str, payload, raw_message: dict) -> bool:
    handler = self.actions.get(action)
    if handler is None:
        logger.error(f"Undefined action: {action}")
        return True
    # ...

@dataclass(frozen=True)
class ActionSpec:
    action: str
    direction: ActionDirection  # "panel_to_node" | "node_to_panel" | "bidirectional" | ...
    critical: bool = False          # 需要 HMAC 签名吗？
    protocol_version: int = 1
    visibility: Literal["public_protocol", "internal"] = "public_protocol"
    safe_flag: str = ""            # 对应 config.toml 里的安全开关名

_bulk(("execute:run_shell",), "panel_to_node", critical=True, safe_flag="execute_command")
_bulk(("terminal:create_session", "terminal:close_session", "terminal:input", "terminal:resize"),
      "panel_to_node", critical=True, safe_flag="connect_terminal")
_bulk(("node:heartbeat", "node:upload_running_data"), "node_to_panel")
_bulk(("docker:image:pull", "docker:container:create", "docker:container:stop"),
      "panel_to_node", critical=True, safe_flag="docker_manager")

def needs_signature(action: str) -> bool:
    spec = _ACTIONS.get(action)
    if spec:
        return spec.critical
    return action.startswith("plugin:")

async def verify_if_required(self, action: str, payload, raw_message: dict,
                              label: str = "关键动作") -> bool:
    if not needs_signature(action):
        return True
    sign_data = raw_message.get("_sign")
    if not sign_data:
        logger.warning(f"{label}缺少签名: {action}")
        await self.client.close_current_ws(code=3001)
        return False
    if not verify_signature(self.client.node_token(), action, payload or {}, sign_data):
        logger.warning(f"{label}签名验证失败: {action}")
        await self.client.close_current_ws(code=3001)
        return False
    return True

# 文件管理类操作
async def file_manager_list(self, data):
    if not self.client.safe_enabled("file_manager", True):
        return
    await self.client.services.file_manager.list_dir(data or {})

from cryptography.fernet import Fernet, InvalidToken
import base64
import hashlib

def _derive_fernet_key(node_token: str) -> bytes:
    """用 node_token 的 SHA-256 派生 Fernet 密钥"""
    digest = hashlib.sha256(node_token.encode('utf-8')).digest()
    return base64.urlsafe_b64encode(digest)

from cryptography.fernet import Fernet
import hashlib
import base64
import json
import time

def issue_terminal_ticket(node_token: str, password: str, ttl: int = 45) -> str:
    key = base64.urlsafe_b64encode(hashlib.sha256(node_token.encode()).digest())
    fernet = Fernet(key)
    payload = json.dumps({
        "password": password,
        "expires_at": time.time() + ttl,
    })
    return fernet.encrypt(payload.encode()).decode()

def decrypt_terminal_password(self, payload: dict | None) -> str:
    payload = payload if isinstance(payload, dict) else {}
    encrypted = payload.get("password_encrypted")
    if not encrypted:
        return ""
    try:
        import base64
        import hashlib

        token = self.client.node_token().encode("utf-8")
        key = base64.urlsafe_b64encode(hashlib.sha256(token).digest())
        raw = Fernet(key).decrypt(str(encrypted).encode("utf-8"), ttl=45)
        data = self.client.decode_json(raw.decode("utf-8"))
        if data.get("expires_at", 0) and float(data["expires_at"]) < self.client.now():
            raise ValueError("终端密码票据已过期")
        return str(data.get("password") or "")
    except (InvalidToken, ValueError, TypeError) as e:
        logger.warning(f"终端密码票据解密失败: {e}")
        return ""

SENSITIVE_LOG_KEYS = {
    'password', 'passwd', 'pwd', 'password_encrypted',
    'token', 'node_token', 'client_token',
    'access', 'refresh', 'authorization',
    'signature', '_sign', 'secret',
}

async def __get_download_url(self, download_task: DownloadTaskConfig) -> tuple[str | None, str | None]:
    headers = {"Authorization": f"Bearer {self.__access_token}"}
    params = {"task": download_task.task_id, "file": download_task.file_id}
    async with self.__session.get(
            self.__url,
            params=params,
            headers=headers,
            timeout=aiohttp.ClientTimeout(total=15)
    ) as response:
        data = await response.json()
        if response.status != 200 or data.get("status") != 1:
            raise RuntimeError(f"下载授权失败({response.status}): {data.get('msg')}")
        payload = data.get("data") or {}
        return payload.get("url"), payload.get("file_name")

class DownloadTaskConfig:
    def __init__(self, task_id: str, file_id: str, file_name: str | None,
                 save_path: str, check_hash: bool, file_hash: str):
        self.task_id = validate_uuid(task_id, "task_id")
        self.file_id = validate_sha256(file_id, "file_id")
        self.file_name = file_name
        self.save_path = save_path
        self.check_hash = check_hash
        self.file_hash = validate_sha256(file_hash or file_id, "file_hash")

def resolve_download_path(self, requested_path: str | None) -> str:
    base = safe_data_dir("download", create=True)
    if self.safe_enabled("allow_custom_download_path", False) and requested_path:
        return str(Path(requested_path).expanduser().resolve())
    if not requested_path or os.path.isabs(str(requested_path)):
        return str(base)
    return str(safe_join(base, requested_path, create_dir=True))

def safe_join(base: str | os.PathLike, *parts: str | os.PathLike, create_dir: bool = False) -> Path:
    base_path = Path(base).resolve()
    target = base_path.joinpath(*[str(part) for part in parts]).resolve()
    if target != base_path and base_path not in target.parents:
        raise SecurityError(f"路径越界: {target}")
    if create_dir:
        target.mkdir(parents=True, exist_ok=True)
    return target

async with self.__session.get(download_url, timeout=aiohttp.ClientTimeout(total=300)) as response:
    if response.status != 200:
        # ... 错误处理
    filename = sanitize_filename(download_task.file_name or authorized_file_name, download_task.file_id)
    target_path = safe_join(save_dir, filename)
    sha256 = hashlib.sha256()
    with open(target_path, 'wb') as file:
        while True:
            chunk = await response.content.read(8192)
            sha256.update(chunk)
            if not chunk:
                break
            file.write(chunk)

def sanitize_filename(filename: Any, fallback: str) -> str:
    fallback = os.path.basename(str(fallback or "download"))
    filename = os.path.basename(str(filename or fallback))
    filename = filename.replace("\x00", "").strip()
    if not filename or filename in {".", ".."}:
        filename = fallback or "download"
    return filename[:255]

sha256 = sha256.hexdigest()
if download_task.check_hash and str(download_task.file_hash) != str(sha256):
    return self.__send_websocket_action('file_download:failure', {
        'task': download_task.task_id,
        'file': download_task.file_id,
        'error_type': "文件哈希校验失败",
        'error_content': f"{download_task.file_hash} != {sha256}"
    })

class DownloadFileUtil:
    __max_download_thread: int = 4
    __download_threads: int = 0
    __thread_lock: Lock()

    def __handle_download_queue(self):
        while True:
            if self.__download_queue.empty():
                break
            if self.__download_threads >= self.__max_download_thread:
                time.sleep(0.3)
                continue
            task = self.__download_queue.get()
            asyncio.run_coroutine_threadsafe(
                self.__download(task), self.__loop
            )
            self.__increment_download_threads()

# 成功
self.__send_websocket_action('file_download:success', {
    'task': download_task.task_id,
    'file': download_task.file_id,
})
# 失败
self.__send_websocket_action('file_download:failure', {
    'task': download_task.task_id,
    'file': download_task.file_id,
    'error_type': "文件哈希校验失败",
    'error_content': f"{download_task.file_hash} != {sha256}"
})

with transaction.atomic():
    order = Order.objects.create(...)
    channel_layer.group_send(f"order_{order.id}", {
        "type": "order.created",
        "order_id": order.id,
    })

class UnitOfWork:
    @contextmanager
    def atomic(self) -> Iterator[None]:
        with transaction.atomic():
            yield

    def on_commit(self, callback: Callable[[], object]) -> None:
        transaction.on_commit(callback)

default_unit_of_work = UnitOfWork()

@dataclass(frozen=True)
class OutboxEvent:
    event_id: str
    event_type: str
    aggregate_type: str
    aggregate_id: str
    idempotency_key: str
    payload: Any
    status: Literal["pending", "sent", "failed"] = "pending"

class OutboxPublisher:
    def __init__(self, *, use_persistent: bool = True):
        self._use_persistent = use_persistent
        self._memory_queue: deque[OutboxEvent] = deque(maxlen=10000)
        self._handlers: dict[str, list[Callable]] = {}

    def publish_after_commit(self, event: OutboxEvent) -> OutboxEvent:
        if self._use_persistent:
            transaction.on_commit(lambda: self._persist_event(event))
        else:
            transaction.on_commit(lambda: self._enqueue_memory(event))
        return event

def _persist_event(self, event: OutboxEvent) -> None:
    try:
        OutboxRecord.objects.update_or_create(
            idempotency_key=event.idempotency_key,
            defaults={
                "event_id": event.event_id,
                "event_type": event.event_type,
                "aggregate_type": event.aggregate_type,
                "aggregate_id": event.aggregate_id,
                "payload": event.payload if isinstance(event.payload, dict) else {"data": str(event.payload)},
                "status": "pending",
            },
        )
    except Exception:
        logger.exception("Failed to persist outbox event: %s", event.event_id)

class OutboxRecord(models.Model):
    event_id = models.CharField(max_length=64, unique=True, primary_key=True)
    event_type = models.CharField(max_length=128, db_index=True)
    aggregate_type = models.CharField(max_length=128)
    aggregate_id = models.CharField(max_length=128, db_index=True)
    idempotency_key = models.CharField(max_length=128, unique=True)
    payload = models.JSONField()
    status = models.CharField(max_length=16, choices=[("pending", "Pending"), ("sent", "Sent"), ("failed", "Failed")], default="pending", db_index=True)
    created_at = models.DateTimeField(auto_now_add=True)
    updated_at = models.DateTimeField(auto_now=True)
    retry_count = models.IntegerField(default=0)

    class Meta:
        db_table = "core_outbox_event"
        indexes = [
            models.Index(fields=["status", "created_at"], name="outbox_status_created_idx"),
        ]

def consume_pending(self, limit: int = 100) -> list[OutboxEvent]:
    if not self._use_persistent:
        return list(self._memory_queue)
    records = list(
        OutboxRecord.objects.filter(status="pending")
        .order_by("created_at")[:limit]
    )
    return [r.to_event() for r in records]

def mark_sent(self, event_id: str) -> None:
    if self._use_persistent:
        OutboxRecord.objects.filter(event_id=event_id).update(status="sent")
    # 内存模式直接从队列里移除
    self._memory_queue = deque(
        (e for e in self._memory_queue if e.event_id != event_id),
        maxlen=self._memory_queue.maxlen,
    )

def mark_failed(self, event_id: str) -> None:
    if self._use_persistent:
        OutboxRecord.objects.filter(event_id=event_id).update(
            status="failed",
            retry_count=models.F("retry_count") + 1,
        )

def retry_failed(self, max_retries: int = 3) -> list[OutboxEvent]:
    if not self._use_persistent:
        return []
    records = list(
        OutboxRecord.objects.filter(
            status="failed",
            retry_count__lt=max_retries,
        ).order_by("created_at")[:100]
    )
    ids = [r.event_id for r in records]
    OutboxRecord.objects.filter(event_id__in=ids).update(status="pending")
    return [r.to_event() for r in records]

def _enqueue_memory(self, event: OutboxEvent) -> None:
    self._memory_queue.append(event)
    self._dispatch_memory(event)

def _dispatch_memory(self, event: OutboxEvent) -> None:
    handlers = self._handlers.get(event.event_type, [])
    for handler in handlers:
        try:
            handler(event)
        except Exception:
            logger.exception("Handler failed for event: %s", event.event_id)

def register_handler(self, event_type: str, handler: Callable[[OutboxEvent], None]) -> None:
    self._handlers.setdefault(event_type, []).append(handler)

from apps.core.persistence.outbox import default_outbox_publisher, OutboxEvent
from apps.core.persistence.unit_of_work import default_unit_of_work
from apps.core.contracts.auth import AuthPrincipal
import uuid
from datetime import datetime

def create_group(name, user_principal: AuthPrincipal):
    with default_unit_of_work.atomic():
        group = Node_Group.objects.create(name=name)

        # 事务内预约事件
        default_outbox_publisher.publish_after_commit(OutboxEvent(
            event_id=str(uuid.uuid4()),
            event_type="group.created",
            occurred_at=datetime.now(),
            actor=user_principal,
            payload={"group_id": group.id, "name": group.name},
            aggregate_type="group",
            aggregate_id=str(group.id),
            idempotency_key=f"group.created:{group.id}",
        ))

    # 事务提交后，OutboxPublisher 才把事件持久化
    # Worker 捞取后发 WebSocket 通知

@dataclass(frozen=True)
class AuthPrincipal:
    principal_type: Literal["user", "node", "download_ticket", "anonymous"]
    user_id: int | None = None
    username: str = ""
    node_uuid: str | None = None
    role_ids: tuple[int, ...] = ()
    permission_codes: tuple[str, ...] = ()
    is_superuser: bool = False
    is_disabled: bool = False

    @property
    def is_node(self) -> bool:
        return self.principal_type == "node"

    def has_permission(self, code: str) -> bool:
        return self.is_superuser or "all" in self.permission_codes or code in self.permission_codes

@dataclass(frozen=True)
class ModuleManifest:
    module_id: str
    label: str
    url_prefix: str = ""
    urls_module: str = ""
    permissions: tuple[PermissionSpec, ...] = ()
    permission_categories: tuple[PermissionCategorySpec, ...] = ()
    routes: tuple[RouteSpec, ...] = ()
    websocket_actions: tuple[WebSocketActionSpec, ...] = ()
    dashboard_providers: tuple[DashboardProviderSpec, ...] = ()
    scheduler_jobs: tuple[SchedulerJobSpec, ...] = ()

class ModuleManifestRegistry:
    def validate(self) -> list[ManifestViolation]:
        violations = []
        seen_modules = set()
        seen_permissions = set()
        seen_routes = set()
        seen_ws = set()
        # ...
        for manifest in self._manifests:
            if manifest.module_id in seen_modules:
                violations.append(ManifestViolation("DUP_MODULE", ...))
            # 权限引用检查：路由声明的权限必须存在于某个模块的 permissions 里
            for route in manifest.routes:
                violations.extend(
                    _validate_requirement(route.permission, all_permissions, ...)
                )
        return violations

class NodeCommandBus:
    def __init__(self, adapter: ChannelsNodeAdapter | None = None):
        self._adapter = adapter or default_channels_node_adapter

    async def send(self, command: NodeCommand) -> NodeCommandResult:
        payload = command.payload
        if is_dataclass(payload):
            payload = asdict(payload)
        event = {
            "type": "send.message",
            "action": command.action,
            "request_id": command.request_id,
            "payload": payload,
            "requires_signature": command.requires_signature,
        }
        await self._adapter.send_to_node(command.node_uuid, event)
        return NodeCommandResult(request_id=command.request_id, status="sent")

class DomainEventBus:
    def __init__(self):
        self._subscribers: dict[str, list[Subscriber]] = defaultdict(list)

    def subscribe(self, event_type: str, subscriber: Subscriber) -> None:
        self._subscribers[event_type].append(subscriber)

    def publish(self, event: DomainEvent) -> list[object]:
        results = []
        for subscriber in tuple(self._subscribers.get(event.event_type, ())):
            results.append(subscriber(event))
        for subscriber in tuple(self._subscribers.get("*", ())):
            results.append(subscriber(event))
        return results

@dataclass(frozen=True)
class DomainEvent:
    event_id: str
    event_type: str
    occurred_at: datetime
    actor: AuthPrincipal | None
    payload: object
    aggregate_type: str = ""
    aggregate_id: str = ""
    idempotency_key: str = ""

class UnitOfWork:
    @contextmanager
    def atomic(self) -> Iterator[None]:
        with transaction.atomic():
            yield

    def on_commit(self, callback: Callable[[], object]) -> None:
        transaction.on_commit(callback)

default_unit_of_work = UnitOfWork()

class OutboxPublisher:
    def publish_after_commit(self, event: OutboxEvent) -> OutboxEvent:
        if self._use_persistent:
            transaction.on_commit(lambda: self._persist_event(event))
        else:
            transaction.on_commit(lambda: self._enqueue_memory(event))
        return event

@dataclass(frozen=True)
class OutboxEvent:
    event_id: str
    event_type: str
    aggregate_type: str
    aggregate_id: str
    idempotency_key: str
    payload: Any
    status: Literal["pending", "sent", "failed"] = "pending"

class ASTValidator:
    BLOCKED_ATTRIBUTE_NAMES = {
        '__class__', '__bases__', '__mro__', '__subclasses__',
        '__globals__', '__code__', '__closure__', '__func__',
        # ... 更多危险属性
    }
    BLOCKED_REFLECTION_FUNCTIONS = {'vars', 'dir', 'globals', 'locals'}

    @staticmethod
    def validate(source_code: str, filepath: str = '',
                 blocked_modules: Optional[Set[str]] = None) -> List[str]:
        tree = ast.parse(source_code)
        warnings = []
        for node in ast.walk(tree):
            if isinstance(node, ast.Import):
                for alias in node.names:
                    if _is_module_blocked(alias.name, blocked_modules):
                        warnings.append(f"{filepath}: 导入被禁止的模块: {alias.name}")
            if isinstance(node, ast.Attribute):
                if node.attr in ASTValidator.BLOCKED_ATTRIBUTE_NAMES:
                    warnings.append(f"{filepath}: 访问被禁止的属性: {node.attr}")
        return warnings

class _PluginImportFinder:
    def find_spec(self, fullname, path, target=None):
        ctx = _plugin_context.get()
        if ctx is None:
            return None  # 非插件上下文，放行

        plugin_id = ctx.get('plugin_id', 'unknown')
        permissions = ctx.get('permissions', [])

        # 关键模块（subprocess, os.system 等）：直接报错
        critical = get_critical_blocked_modules()
        if _is_module_blocked(fullname, critical):
            raise ImportError(f"插件无权导入模块: {fullname}")

        # 根据权限列表过滤
        blocked = get_blocked_modules_for_permissions(permissions)
        if _is_module_blocked(fullname, blocked):
            raise ImportError(f"插件无权导入被禁止的模块: {fullname}")

        # 内部模块（apps.*, django.* 等）：需要走白名单
        root_module = fullname.split('.')[0]
        if root_module in ('apps', 'middleware', 'consumers', 'util', 'django'):
            if not self._is_safe_internal_import(fullname, permissions):
                raise ImportError(f"插件无权直接导入: {fullname}")

        return None  # 放行

@contextlib.contextmanager
def runtime_guard():
    patches = []

    def patch(obj, attr, replacement):
        original = getattr(obj, attr)
        patches.append((obj, attr, original))
        setattr(obj, attr, replacement)

    # 直接封杀的：eval、exec、compile、breakpoint
    for name in ('eval', 'exec', 'compile', 'breakpoint'):
        patch(builtins, name, _blocked_runtime_call_factory(getattr(builtins, name), name))

    # 属性访问守卫：getattr、setattr、delattr
    patch(builtins, 'getattr', _guarded_getattr_factory(builtins.getattr))
    patch(builtins, 'setattr', _guarded_setattr_factory(builtins.setattr))
    patch(builtins, 'delattr', _guarded_delattr_factory(builtins.delattr))

    # 反射函数封杀：globals、locals、vars、dir
    for name in ('globals', 'locals', 'vars', 'dir'):
        patch(builtins, name, _guarded_reflection_call_factory(getattr(builtins, name), name))

    try:
        yield
    finally:
        # 还原——从后往前还原，确保嵌套时不出问题
        for obj, attr, original in reversed(patches):
            setattr(obj, attr, original)

def _blocked_runtime_call_factory(original_func, name: str):
    def _inner(*args, **kwargs):
        # 框架代码信任标记为 True，且调用栈顶部不是插件代码——放行
        if _trusted_internal_import.get() and not _called_from_plugin_frame(caller_globals):
            return original_func(*args, **kwargs)
        # 否则直接报错
        raise PermissionError(f"插件 {plugin_id} 无权调用危险函数: {name}")
    return _inner

BLOCKED_ATTRIBUTE_NAMES = {
    '__class__', '__bases__', '__mro__', '__subclasses__',
    '__globals__', '__code__', '__closure__', '__func__',
    '__self__', '__module__', '__dict__', '__defaults__',
    'gi_frame', 'gi_code', 'cr_frame', 'cr_code', 'tb_frame',
    'f_globals', 'f_locals', 'func_globals',
    'modules', 'path_hooks', 'meta_path',
}

def plugin_guarded_getattr(obj, name):
    if name in BLOCKED_ATTRIBUTE_NAMES:
        raise AttributeError(f'"_getattr_" not allowed to access dangerous name "{name}"')
    if name.startswith('_') and name not in ALLOWED_DUNDER_METHODS:
        raise AttributeError(f'"_getattr_" not allowed to access private name "{name}"')
    return getattr(obj, name)

def _guarded_os_open_factory(original_open):
    def _guarded_os_open(path, flags, mode=0o777, *args, **kwargs):
        ctx = _plugin_context.get()
        if ctx is None:
            return original_open(path, flags, mode, *args, **kwargs)
        if kwargs.get('dir_fd') is not None:
            plugin_id = ctx.get('plugin_id', 'unknown')
            raise PermissionError(f"插件 {plugin_id} 无权通过 dir_fd 访问文件")
        # ... 黑白名单检查
    return _guarded_os_open

def _is_path_inside(base_path: str, candidate_path: str) -> bool:
    base_real = os.path.realpath(base_path)
    candidate_real = os.path.realpath(candidate_path)
    return os.path.commonpath([base_real, candidate_real]) == base_real

_plugin_context: ContextVar = ContextVar('plugin_context', default=None)
_policy_path_resolution: ContextVar = ContextVar('policy_path_resolution', default=False)
_trusted_internal_import: ContextVar = ContextVar('trusted_internal_import', default=False)
_third_party_caller: ContextVar = ContextVar('_third_party_caller', default=False)

async def execute_safe_async(self, plugin_id, func, *args, ...):
    ctx = copy_context()
    # 同步函数放进线程池，带着自己的上下文跑
    result = await asyncio.wait_for(
        loop.run_in_executor(None, lambda: ctx.run(func, *args, **kwargs)),
        timeout=timeout
    )

async def write_node_usage(cls, node_uuid, cpu_usage, memory_used, ...):
    node_uuid = _validate_node_uuid(node_uuid)
    await ensure_node_subtables(node_uuid)  # 确保子表存在
    uuid_safe = str(node_uuid).replace('-', '_')
    subtable = f"{TD_SUBTABLE_NODE_USAGE}_{uuid_safe}"
    ts = int(timestamp.timestamp() * 1000) if timestamp else "NOW"
    sql = (
        f"INSERT INTO {subtable} VALUES "
        f"({ts}, {cpu_usage}, {memory_used}, {swap_used}, "
        f"{disk_io_read}, {disk_io_write}, "
        f"{loadavg_1m}, {loadavg_5m}, {loadavg_15m})"
    )
    return await tdengine_client.ainsert(sql)

async def write_node_usage_batch(cls, node_uuid, cpu_usage, memory_used, ...,
                                  cpu_cores, network_io, accelerators=None,
                                  timestamp=None):
    uuid_safe = str(node_uuid).replace('-', '_')
    ts = int(timestamp.timestamp() * 1000) if timestamp else "NOW"

    results = []

    # 1. 整机使用率
    subtable = f"{TD_SUBTABLE_NODE_USAGE}_{uuid_safe}"
    sql = f"INSERT INTO {subtable} VALUES ({ts}, {cpu_usage}, ...)"
    results.append(await tdengine_client.ainsert(sql))

    # 2. 每核 CPU —— 合并成一条 INSERT
    if cpu_cores:
        core_inserts = []
        for core_data in cpu_cores:
            await ensure_cpu_core_subtable(node_uuid, core_data['core_index'])
            core_subtable = f"{TD_SUBTABLE_NODE_CPU_CORE}_{uuid_safe}_{core_data['core_index']}"
            core_inserts.append(f"{core_subtable} VALUES ({ts}, {core_data['usage']})")
        sql = "INSERT INTO " + " ".join(core_inserts)
        results.append(await tdengine_client.ainsert(sql))

    # 3. 网络流量 —— 同理合并
    if network_io:
        net_inserts = []
        for port_name, io_data in network_io.items():
            await ensure_network_subtable(node_uuid, port_name)
            port_safe = port_name.replace('.', '_').replace('-', '_')[:40]
            net_subtable = f"{TD_SUBTABLE_NODE_NETWORK}_{uuid_safe}_{port_safe}"
            net_inserts.append(f"{net_subtable} VALUES ({ts}, {io_data['bytes_sent']}, {io_data['bytes_recv']})")
        sql = "INSERT INTO " + " ".join(net_inserts)
        results.append(await tdengine_client.ainsert(sql))

    # 4. 加速器（GPU/NPU）
    if accelerators:
        # ... 同理

    return all(results) if results else True

_UUID_RE = re.compile(r'^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$', re.IGNORECASE)
_PORT_NAME_RE = re.compile(r'^[A-Za-z0-9_.\-]{1,40}$')

def _validate_node_uuid(node_uuid: str) -> str:
    uuid_str = str(node_uuid or '').strip()
    if not _UUID_RE.match(uuid_str):
        raise ValueError(f"无效的 node_uuid 格式: {uuid_str}")
    return uuid_str

def _validate_port_name(port_name: str) -> str:
    name = str(port_name or '').strip()
    if not _PORT_NAME_RE.match(name):
        raise ValueError(f"无效的网卡名: {name}")
    return name

@staticmethod
async def get_last_node_usage(node_uuid: str) -> Optional[dict]:
    # 先查 Redis
    # Redis 层没在 query.py 里，而是在 Node Manager 的 cache 里
    # 这里是兜底：Redis 没命中时查 TDengine
    uuid_safe = str(node_uuid).replace('-', '_')
    subtable = f"{TD_SUBTABLE_NODE_USAGE}_{uuid_safe}"
    sql = f"SELECT LAST_ROW(*) FROM {subtable}"
    try:
        rows = await tdengine_client.aquery(sql)
        if rows:
            row = rows[0]
            row['timestamp'] = _ms_to_datetime(row.get('ts', 0))
            return row
    except Exception:
        pass
    return None

def _parse_to_ms(time_str: str) -> int:
    for fmt in ('%Y-%m-%d %H:%M:%S', '%Y-%m-%dT%H:%M', '%Y-%m-%dT%H:%M:%S', '%Y-%m-%d'):
        try:
            dt = datetime.strptime(time_str, fmt)
            return int(dt.timestamp() * 1000)
        except ValueError:
            continue
    return int(datetime.now().timestamp() * 1000)

@staticmethod
async def query_performance_record(node_uuid, start_time, end_time,
                                    device="_all", processor_count=0, memory_total=0):
    usage_rows = await TDengineQuery.query_node_usage(node_uuid, start_time, end_time)
    if not usage_rows:
        return []

    # 按设备类型筛选
    cpu_core_map = {}   # {ts: [{core, usage}, ...]}
    net_map = {}        # {ts: [{name, bytes_sent, bytes_recv}, ...]}
    accelerator_map = {}

    if device in ("_all", "cpu"):
        cpu_core_rows = await TDengineQuery.query_cpu_core_usage(...)
        for r in cpu_core_rows:
            cpu_core_map.setdefault(r.get('ts', 0), []).append({...})

    # ... network, accelerator 同理

    result = []
    for row in usage_rows:
        ts = row.get('ts', 0)
        item = {"timestamp": row['timestamp']}
        if device == "_all" or device == "cpu":
            item["cpu_usage"] = row.get('cpu_usage', 0)
            item["cpu_cores_usage"] = cpu_core_map.get(ts, [])
        if device == "_all" or device == "network":
            item["network_usage"] = net_map.get(ts, [])
        # ...
        result.append(item)
    return result