agent-Specialization/server/chat/permission.py

from __future__ import annotations
from server.chat import chat_bp
from server.chat.settings import _dispatch_runtime_mode_notice
import json, time

PERMISSION_MODE_OPTIONS = ["readonly", "approval", "auto_approval", "unrestricted"]
EXECUTION_MODE_OPTIONS = ["sandbox", "direct"]
NETWORK_PERMISSION_OPTIONS = ["restricted", "full", "none"]
from datetime import datetime
from typing import Dict, Any, Optional
from pathlib import Path
from io import BytesIO
import zipfile
import os

from flask import Blueprint, jsonify, request, session, send_file
from werkzeug.utils import secure_filename
from werkzeug.exceptions import RequestEntityTooLarge
import secrets

from config import THINKING_FAST_INTERVAL, MAX_UPLOAD_SIZE, OUTPUT_FORMATS
from modules.personalization_manager import (
    load_personalization_config,
    resolve_context_compression_settings,
    save_personalization_config,
    THINKING_INTERVAL_MIN,
    THINKING_INTERVAL_MAX,
    RECENT_CONVERSATIONS_PROMPT_LIMIT_MIN,
    RECENT_CONVERSATIONS_PROMPT_LIMIT_MAX,
)
from modules.skills_manager import (
    get_skills_catalog,
    infer_private_skills_dir,
    merge_enabled_skills,
    sync_workspace_skills,
)
from modules.upload_security import UploadSecurityError
from modules.host_sandbox_policy import load_policy, save_policy
from modules.user_manager import UserWorkspace
from core.web_terminal import WebTerminal
from config.model_profiles import get_model_context_window

from server.auth_helpers import api_login_required, resolve_admin_policy, get_current_user_record, get_current_username
from server.context import with_terminal, get_gui_manager, get_upload_guard, build_upload_error_response, ensure_conversation_loaded, get_or_create_usage_tracker
from server.security import rate_limited, prune_socket_tokens
from server.utils_common import debug_log
from server.state import PROJECT_MAX_STORAGE_MB, THINKING_FAILURE_KEYWORDS, pending_socket_tokens, SOCKET_TOKEN_TTL_SECONDS
from server.state import tool_approval_manager, user_question_manager
from server.extensions import socketio
from server.monitor import get_cached_monitor_snapshot
from server.files import sanitize_filename_preserve_unicode

UPLOAD_FOLDER_NAME = ".agents/user_upload"
@chat_bp.route('/api/permission-mode', methods=['GET'])
@api_login_required
@with_terminal
def get_permission_mode(terminal: WebTerminal, workspace: UserWorkspace, username: str):
    """获取当前权限模式。"""
    current_conversation_id = getattr(terminal.context_manager, "current_conversation_id", None)
    return jsonify({
        "success": True,
        "mode": terminal.get_permission_mode() if hasattr(terminal, "get_permission_mode") else "unrestricted",
        "pending_mode": (terminal.get_pending_runtime_modes().get("permission_mode") if hasattr(terminal, "get_pending_runtime_modes") else None),
        "options": PERMISSION_MODE_OPTIONS,
        "conversation_id": current_conversation_id,
    })

@chat_bp.route('/api/permission-mode', methods=['POST'])
@api_login_required
@with_terminal
@rate_limited("permission_mode_switch", 30, 60, scope="user")
def update_permission_mode(terminal: WebTerminal, workspace: UserWorkspace, username: str):
    """更新当前对话权限模式。"""
    data = request.get_json() or {}
    target_mode = str(data.get("mode") or "").strip().lower()
    if target_mode not in PERMISSION_MODE_OPTIONS:
        return jsonify({
            "success": False,
            "error": "无效权限模式，仅支持 readonly / approval / auto_approval / unrestricted"
        }), 400

    # 判断当前是否在对话运行期间
    is_running = False
    try:
        from server.tasks import task_manager
        current_conv = getattr(getattr(terminal, "context_manager", None), "current_conversation_id", None)
        running_tasks = [
            r for r in task_manager.list_tasks(username) if r.status in {"pending", "running", "cancel_requested"}
        ]
        if current_conv:
            running_tasks.sort(key=lambda r: 0 if r.conversation_id == current_conv else 1)
        is_running = bool(running_tasks)
    except Exception:
        pass

    if is_running:
        # 运行期间：使用 pending 机制延迟到工具循环中统一处理，
        # 避免「A→B→A」来回切换时中间值被覆盖后仍插入多余消息。
        try:
            terminal.queue_permission_mode_change(target_mode)
        except Exception as exc:
            return jsonify({
                "success": False,
                "error": str(exc),
                "message": "更新权限模式失败"
            }), 500
        status = terminal.get_status()
        socketio.emit('status_update', status, room=f"user_{username}")
        return jsonify({
            "success": True,
            "mode": target_mode,
            "pending_mode": target_mode,
            "options": PERMISSION_MODE_OPTIONS,
            "conversation_id": getattr(terminal.context_manager, "current_conversation_id", None),
            "message": "权限模式将在当前工具执行完成后生效",
        })

    # 空闲期间：直接生效
    previous_mode = terminal.get_permission_mode() if hasattr(terminal, "get_permission_mode") else None
    try:
        applied_mode = terminal.set_permission_mode(target_mode)
        if hasattr(terminal, "pending_permission_mode"):
            terminal.pending_permission_mode = None
        if hasattr(terminal, "_persist_runtime_mode_metadata"):
            terminal._persist_runtime_mode_metadata({
                "permission_mode": applied_mode,
                "pending_permission_mode": None,
            })
    except Exception as exc:
        return jsonify({
            "success": False,
            "error": str(exc),
            "message": "更新权限模式失败"
        }), 500

    session["permission_mode"] = applied_mode
    if applied_mode != previous_mode:
        permission_label = {
            "readonly": "只读",
            "approval": "批准",
            "auto_approval": "自动审核",
            "unrestricted": "无限制",
        }.get(applied_mode, applied_mode)
        _dispatch_runtime_mode_notice(terminal, username, f"权限模式被用户修改为 {permission_label}", source="权限变更")
    status = terminal.get_status()
    socketio.emit('status_update', status, room=f"user_{username}")
    return jsonify({
        "success": True,
        "mode": applied_mode,
        "pending_mode": None,
        "options": PERMISSION_MODE_OPTIONS,
        "conversation_id": getattr(terminal.context_manager, "current_conversation_id", None),
        "message": "权限模式已更新并立即生效",
    })

@chat_bp.route('/api/execution-mode', methods=['GET'])
@api_login_required
@with_terminal
def get_execution_mode(terminal: WebTerminal, workspace: UserWorkspace, username: str):
    is_host = bool(getattr(terminal, "_is_host_mode", lambda: False)())
    can_manage = is_host and getattr(terminal, "user_role", "user") == "admin"
    state = terminal.get_execution_mode_state() if hasattr(terminal, "get_execution_mode_state") else {"mode": "sandbox"}
    return jsonify({
        "success": True,
        "enabled": can_manage,
        "state": state,
        "pending_mode": (terminal.get_pending_runtime_modes().get("execution_mode") if hasattr(terminal, "get_pending_runtime_modes") else None),
        "options": EXECUTION_MODE_OPTIONS,
    })

@chat_bp.route('/api/execution-mode', methods=['POST'])
@api_login_required
@with_terminal
@rate_limited("execution_mode_switch", 20, 60, scope="user")
def update_execution_mode(terminal: WebTerminal, workspace: UserWorkspace, username: str):
    is_host = bool(getattr(terminal, "_is_host_mode", lambda: False)())
    can_manage = is_host and getattr(terminal, "user_role", "user") == "admin"
    if not can_manage:
        return jsonify({"success": False, "error": "仅宿主机管理员可切换执行环境"}), 403
    data = request.get_json() or {}
    target_mode = str(data.get("mode") or "").strip().lower()
    if target_mode not in EXECUTION_MODE_OPTIONS:
        return jsonify({"success": False, "error": "无效执行环境，仅支持 sandbox / direct"}), 400

    # 判断当前是否在对话运行期间
    is_running = False
    try:
        from server.tasks import task_manager
        current_conv = getattr(getattr(terminal, "context_manager", None), "current_conversation_id", None)
        running_tasks = [
            r for r in task_manager.list_tasks(username) if r.status in {"pending", "running", "cancel_requested"}
        ]
        if current_conv:
            running_tasks.sort(key=lambda r: 0 if r.conversation_id == current_conv else 1)
        is_running = bool(running_tasks)
    except Exception:
        pass

    if is_running:
        # 运行期间：使用 pending 机制延迟到工具循环中统一处理，
        # 避免「A→B→A」来回切换时中间值被覆盖后仍插入多余消息。
        try:
            terminal.queue_execution_mode_change(target_mode)
        except Exception as exc:
            return jsonify({"success": False, "error": str(exc), "message": "更新执行环境失败"}), 500
        status = terminal.get_status()
        socketio.emit('status_update', status, room=f"user_{username}")
        return jsonify({
            "success": True,
            "state": {
                **(terminal.get_execution_mode_state() if hasattr(terminal, "get_execution_mode_state") else {"mode": target_mode}),
                "mode": target_mode,
            },
            "pending_mode": target_mode,
            "options": EXECUTION_MODE_OPTIONS,
            "message": "执行环境将在当前工具执行完成后生效",
        })

    # 空闲期间：直接生效
    previous_mode = terminal.get_execution_mode() if hasattr(terminal, "get_execution_mode") else None
    try:
        state = terminal.set_execution_mode(target_mode)
        if hasattr(terminal, "pending_execution_mode"):
            terminal.pending_execution_mode = None
        if hasattr(terminal, "_persist_runtime_mode_metadata"):
            terminal._persist_runtime_mode_metadata({
                "execution_mode": state.get("mode", target_mode),
                "pending_execution_mode": None,
            })
    except Exception as exc:
        return jsonify({"success": False, "error": str(exc), "message": "更新执行环境失败"}), 500
    status = terminal.get_status()
    current_mode = state.get("mode", target_mode)
    if current_mode != previous_mode:
        execution_label = {"sandbox": "沙箱", "direct": "完全访问权限"}.get(current_mode, current_mode)
        _dispatch_runtime_mode_notice(terminal, username, f"执行环境被用户修改为 {execution_label}", source="执行环境变更")
    socketio.emit('status_update', status, room=f"user_{username}")
    return jsonify({
        "success": True,
        "state": state,
        "pending_mode": None,
        "options": EXECUTION_MODE_OPTIONS,
        "message": "执行环境已更新并立即生效",
    })

@chat_bp.route('/api/network-permission', methods=['GET'])
@api_login_required
@with_terminal
def get_network_permission(terminal: WebTerminal, workspace: UserWorkspace, username: str):
    is_host = bool(getattr(terminal, "_is_host_mode", lambda: False)())
    can_manage = is_host and getattr(terminal, "user_role", "user") == "admin"
    current = terminal.get_network_permission() if hasattr(terminal, "get_network_permission") else "restricted"
    return jsonify({
        "success": True,
        "enabled": can_manage,
        "mode": current,
        "pending_mode": (terminal.get_pending_runtime_modes().get("network_permission") if hasattr(terminal, "get_pending_runtime_modes") else None),
        "options": NETWORK_PERMISSION_OPTIONS,
    })

@chat_bp.route('/api/network-permission', methods=['POST'])
@api_login_required
@with_terminal
@rate_limited("network_permission_switch", 20, 60, scope="user")
def update_network_permission(terminal: WebTerminal, workspace: UserWorkspace, username: str):
    is_host = bool(getattr(terminal, "_is_host_mode", lambda: False)())
    can_manage = is_host and getattr(terminal, "user_role", "user") == "admin"
    if not can_manage:
        return jsonify({"success": False, "error": "仅宿主机管理员可切换网络权限"}), 403
    data = request.get_json() or {}
    target_mode = str(data.get("mode") or "").strip().lower()
    if target_mode not in NETWORK_PERMISSION_OPTIONS:
        return jsonify({"success": False, "error": "无效网络权限，仅支持 restricted / full"}), 400

    is_running = False
    try:
        from server.tasks import task_manager
        current_conv = getattr(getattr(terminal, "context_manager", None), "current_conversation_id", None)
        running_tasks = [
            r for r in task_manager.list_tasks(username) if r.status in {"pending", "running", "cancel_requested"}
        ]
        if current_conv:
            running_tasks.sort(key=lambda r: 0 if r.conversation_id == current_conv else 1)
        is_running = bool(running_tasks)
    except Exception:
        pass

    if is_running:
        try:
            terminal.queue_network_permission_change(target_mode)
        except Exception as exc:
            return jsonify({"success": False, "error": str(exc), "message": "更新网络权限失败"}), 500
        status = terminal.get_status()
        socketio.emit('status_update', status, room=f"user_{username}")
        return jsonify({
            "success": True,
            "mode": target_mode,
            "pending_mode": target_mode,
            "options": NETWORK_PERMISSION_OPTIONS,
            "message": "网络权限将在当前工具执行完成后生效",
        })

    previous_mode = terminal.get_network_permission() if hasattr(terminal, "get_network_permission") else None
    try:
        applied = terminal.set_network_permission(target_mode)
        if hasattr(terminal, "pending_network_permission"):
            terminal.pending_network_permission = None
        if hasattr(terminal, "_persist_runtime_mode_metadata"):
            terminal._persist_runtime_mode_metadata({
                "network_permission": applied,
                "pending_network_permission": None,
            })
    except Exception as exc:
        return jsonify({"success": False, "error": str(exc), "message": "更新网络权限失败"}), 500
    status = terminal.get_status()
    if applied != previous_mode:
        label = {"restricted": "受限", "full": "完全开放", "none": "完全禁止"}.get(applied, applied)
        _dispatch_runtime_mode_notice(terminal, username, f"网络权限被用户修改为 {label}", source="网络权限变更")
    socketio.emit('status_update', status, room=f"user_{username}")
    return jsonify({
        "success": True,
        "mode": applied,
        "pending_mode": None,
        "options": NETWORK_PERMISSION_OPTIONS,
        "message": "网络权限已更新并立即生效",
    })

@chat_bp.route('/api/path-authorization', methods=['GET'])
@api_login_required
@with_terminal
def get_path_authorization(terminal: WebTerminal, workspace: UserWorkspace, username: str):
    is_host = bool(getattr(terminal, "_is_host_mode", lambda: False)())
    can_manage = is_host and getattr(terminal, "user_role", "user") == "admin"
    data = load_policy()
    return jsonify({
        "success": True,
        "enabled": can_manage,
        "writable_paths": data.get("macos_writable_paths", []),
        "readable_extra_paths": data.get("macos_readable_extra_paths", []),
    })

@chat_bp.route('/api/path-authorization', methods=['POST'])
@api_login_required
@with_terminal
@rate_limited("path_authorization_update", 20, 60, scope="user")
def update_path_authorization(terminal: WebTerminal, workspace: UserWorkspace, username: str):
    is_host = bool(getattr(terminal, "_is_host_mode", lambda: False)())
    can_manage = is_host and getattr(terminal, "user_role", "user") == "admin"
    if not can_manage:
        return jsonify({"success": False, "error": "仅宿主机管理员可管理路径授权"}), 403
    data = request.get_json() or {}
    writable_items = data.get("writable_paths")
    readable_items = data.get("readable_extra_paths")
    if not isinstance(writable_items, list) or not isinstance(readable_items, list):
        return jsonify({"success": False, "error": "writable_paths/readable_extra_paths 必须为数组"}), 400
    writable = [str(x).strip() for x in writable_items if str(x).strip()]
    readable_extra = [str(x).strip() for x in readable_items if str(x).strip()]
    if "/" in writable or "/" in readable_extra:
        return jsonify({"success": False, "error": "禁止授权根目录 /"}), 400
    blocked_prefix = ("~/.ssh", "~/.gnupg", "~/.aws")
    for p in writable + readable_extra:
        lowered = p.lower()
        if any(lowered.startswith(prefix) for prefix in blocked_prefix):
            return jsonify({"success": False, "error": f"禁止授权敏感目录: {p}"}), 400
    payload = save_policy({
        "macos_writable_paths": writable,
        "macos_readable_extra_paths": readable_extra
    })
    return jsonify({
        "success": True,
        "writable_paths": payload.get("macos_writable_paths", []),
        "readable_extra_paths": payload.get("macos_readable_extra_paths", []),
    })