Upload 8 files

app.py

@@ -71,10 +71,11 @@ from router_items import (router as items_router, start_resource_reads, stop_res
 from router_comments import (router as comments_router, start_comment_writes, stop_comment_writes, close_comment_writes) # 💬 评论系统
 from router_messages import (router as messages_router, start_message_io, stop_message_io, close_message_io) # ✉️ 私信系统
 from router_wallet import (router as wallet_router, start_wallet_sql, stop_wallet_sql, close_wallet_sql)     # 💰 钱包/提现
+from proxy_access import start_proxy_access, stop_proxy_access, close_proxy_access
 from router_proxy import router as proxy_router       # 🔗 代理下载
 from router_posts import (router as posts_router, start_posts_io, stop_posts_io, close_posts_io)       # 💬 讨论区
 from router_prompts import (router as prompts_router, start_prompts_io, stop_prompts_io, close_prompts_io)   # ✨ 提示词市场
-from router_tasks import router as tasks_router       # 📝 任务榜
+from router_tasks import (router as tasks_router, start_tasks_io, stop_tasks_io, close_tasks_io)       # 📝 任务榜
 from router_open_api import start_open_sql, stop_open_sql, close_open_sql
 from router_open_api import router as open_api_router  # 🌐 通用开放API（第三方插件）
 from router_admin_plugins import router as admin_plugins_router  # 🔒 管理后台 - 第三方插件管理API
@@ -301,6 +302,8 @@ async def on_startup():
             start_message_io()
             start_posts_io()
             start_prompts_io()
+            start_tasks_io()
+            start_proxy_access()
             start_comment_reads()
             start_admin_scripts()
 
@@ -361,7 +364,7 @@ async def _shutdown_resources():
     for stage, stop in (
         ('stop_wallet_sql', stop_wallet_sql),
         ('stop_resource_reads', stop_resource_reads), ('stop_comment_writes', stop_comment_writes),
-        ('stop_posts', stop_posts_io), ('stop_prompts', stop_prompts_io),
+        ('stop_posts', stop_posts_io), ('stop_prompts', stop_prompts_io), ('stop_tasks', stop_tasks_io), ('stop_proxy_access', stop_proxy_access),
         ('stop_messages', stop_message_io), ('stop_resources', stop_resource_work), ('stop_open_sql', stop_open_sql), ('stop_shared', stop_shared_work),
         ('stop_identity', stop_identity_io), ('stop_auth', stop_auth_io),
         ('stop_verification', stop_verification_io), ('stop_scripts', stop_admin_scripts),
@@ -378,7 +381,7 @@ async def _shutdown_resources():
         ('wait_warmup', wait_startup_warmup, (app.state,)),
         ('close_wallet_sql', close_wallet_sql, ()),
         ('close_resource_reads', close_resource_reads, ()), ('close_comment_writes', close_comment_writes, ()),
-        ('close_posts', close_posts_io, ()), ('close_prompts', close_prompts_io, ()),
+        ('close_posts', close_posts_io, ()), ('close_prompts', close_prompts_io, ()), ('close_tasks', close_tasks_io, ()), ('close_proxy_access', close_proxy_access, ()),
         ('close_messages', close_message_io, ()), ('close_resources', close_resource_work, ()), ('close_open_sql', close_open_sql, ()), ('close_shared', close_shared_work, ()),
         ('close_identity', close_identity_io, ()), ('close_auth', close_auth_io, ()),
         ('close_verification', close_verification_io, ()), ('close_profile', close_profile_io, ()),

proxy_access.py

@@ -1,62 +1,82 @@
-"""下载代理共用资源/有效价格/当前所有权核查，不把存储故障视为免费。"""
-import asyncio
-from models_sql import Ownership
-from purchase_transaction import effective_resource_price
-
-
-class ProxyAccessError(Exception):
-    def __init__(self, status_code, message):
-        self.status_code = status_code
-        self.message = message
-        super().__init__(message)
-
-
-def read_proxy_item(store, item_id):
-    try:
-        source = store.load_data('items.json', default_data=[], strict=True, skip_cache=True)
-        if not isinstance(source, list):
-            raise ValueError('invalid resource source')
-        if any(not isinstance(item, dict) or not isinstance(item.get('id'), str) or not item['id'] for item in source):
-            raise ValueError('invalid resource record')
-        matches = [item for item in source if item.get('id') == item_id]
-        if not matches:
-            raise ProxyAccessError(404, '资源不存在')
-        if len(matches) != 1:
-            raise ValueError('ambiguous resource')
-        item = matches[0]
-        author = item.get('author')
-        if not isinstance(author, str) or not author.strip():
-            raise ValueError('invalid resource author')
-        effective_resource_price(item)
-        return item
-    except ProxyAccessError:
-        raise
-    except Exception:
-        raise ProxyAccessError(503, '资源信息暂时无法核实，请稍后核对') from None
-
-
-async def authorized_proxy_item(store, session, item_id, account):
-    try:
-        item = await asyncio.to_thread(read_proxy_item, store, item_id)
-        author = item['author']
-        price = effective_resource_price(item)
-        if price > 0 and account != author:
-            # SQL会话仍留在调用线程；不用陈旧的身份/拥有缓存，也不授予已退款记录。
-            ownerships = session.query(Ownership).filter(Ownership.account == account, Ownership.item_id == item_id).populate_existing().all()
-            if any(type(row.is_refunded) is not bool for row in ownerships):
-                raise ValueError('invalid ownership refund state')
-            rows = [row for row in ownerships if not row.is_refunded]
-            if len(rows) > 1:
-                raise ValueError('ambiguous ownership')
-            if not rows:
-                raise ProxyAccessError(403, '尚未获取该资源')
-        return item
-    except ProxyAccessError:
-        raise
-    except Exception:
-        raise ProxyAccessError(503, '下载授权暂时无法核实，请稍后核对') from None
-    finally:
-        try:
-            session.rollback()  # 不让下载流长期持有所有权读取事务。
-        except Exception:
-            raise ProxyAccessError(503, '下载授权暂时无法核实，请稍后核对') from None
+"""下载代理共用资源/有效价格/当前所有权核查，不把存储故障视为免费。"""
+from owned_thread import OwnedThreadGroup
+from models_sql import Ownership
+from purchase_transaction import effective_resource_price
+
+
+class ProxyAccessError(Exception):
+    def __init__(self, status_code, message):
+        self.status_code = status_code
+        self.message = message
+        super().__init__(message)
+
+
+def read_proxy_item(store, item_id):
+    try:
+        source = store.load_data('items.json', default_data=[], strict=True, skip_cache=True)
+        if not isinstance(source, list):
+            raise ValueError('invalid resource source')
+        if any(not isinstance(item, dict) or not isinstance(item.get('id'), str) or not item['id'] for item in source):
+            raise ValueError('invalid resource record')
+        matches = [item for item in source if item.get('id') == item_id]
+        if not matches:
+            raise ProxyAccessError(404, '资源不存在')
+        if len(matches) != 1:
+            raise ValueError('ambiguous resource')
+        item = matches[0]
+        author = item.get('author')
+        if not isinstance(author, str) or not author.strip():
+            raise ValueError('invalid resource author')
+        effective_resource_price(item)
+        return item
+    except ProxyAccessError:
+        raise
+    except Exception:
+        raise ProxyAccessError(503, '资源信息暂时无法核实，请稍后核对') from None
+
+
+def _authorized_proxy_item(store, session, item_id, account):
+    try:
+        item = read_proxy_item(store, item_id)
+        author = item['author']
+        price = effective_resource_price(item)
+        if price > 0 and account != author:
+            # SQL会话仍留在调用线程；不用陈旧的身份/拥有缓存，也不授予已退款记录。
+            ownerships = session.query(Ownership).filter(Ownership.account == account, Ownership.item_id == item_id).populate_existing().all()
+            if any(type(row.is_refunded) is not bool for row in ownerships):
+                raise ValueError('invalid ownership refund state')
+            rows = [row for row in ownerships if not row.is_refunded]
+            if len(rows) > 1:
+                raise ValueError('ambiguous ownership')
+            if not rows:
+                raise ProxyAccessError(403, '尚未获取该资源')
+        return item
+    except ProxyAccessError:
+        raise
+    except Exception:
+        raise ProxyAccessError(503, '下载授权暂时无法核实，请稍后核对') from None
+    finally:
+        try:
+            session.rollback()  # 不让下载流长期持有所有权读取事务。
+        except Exception:
+            raise ProxyAccessError(503, '下载授权暂时无法核实，请稍后核对') from None
+
+
+MAX_PROXY_ACCESS = 8
+_proxy_access_io = OwnedThreadGroup(MAX_PROXY_ACCESS,
+    lambda: ProxyAccessError(503, '下载授权暂时无法核实，请稍后核对'))
+start_proxy_access = _proxy_access_io.start
+stop_proxy_access = _proxy_access_io.stop
+close_proxy_access = _proxy_access_io.close
+
+
+async def authorized_proxy_item(store, session_factory, item_id, account):
+    def execute():
+        try:
+            with session_factory() as session:
+                return _authorized_proxy_item(store, session, item_id, account)
+        except ProxyAccessError:
+            raise
+        except Exception:
+            raise ProxyAccessError(503, '下载授权暂时无法核实，请稍后核对') from None
+    return await _proxy_access_io.run(execute)

router_proxy.py

@@ -1,206 +1,205 @@
-# router_proxy.py
-from fastapi import APIRouter, Depends, HTTPException, Request
-from rate_limit import shared_limit
-from fastapi.responses import StreamingResponse, JSONResponse, Response
-from sqlalchemy.orm import Session
-from pydantic import BaseModel, Field, field_validator, HttpUrl
-from typing import Optional
-from urllib.parse import urlparse, urljoin
-import ipaddress
-import httpx
-import os
-import re
-import urllib.request
-import urllib.error
-import 数据库连接 as json_db
-from database_sql import get_db
-from models_sql import Ownership
-from workflow_source import authorized_workflow_path, workflow_credential_target, WorkflowSourceError
-from proxy_access import authorized_proxy_item, ProxyAccessError
-from github_source import repository_parts, select_github_token
-from github_archive import ARCHIVE_MAX_BYTES
-from 安全认证 import require_auth, JWT_SECRET
-from safe_http_transport import ValidatedAsyncHTTPTransport, _resolve_public_ips, PublicResolutionError
-from workflow_download_service import prepare_workflow_download
-from zip_stream import ZIP_DOWNLOAD_MAX_BYTES
-from github_zip_service import prepare_github_zip
-from proxy_resources import close_redirect_response
-from skill_archive_source import install_source, SkillSourceError
-
-router = APIRouter()
-MAX_PROXY_DOWNLOAD_BYTES = int(os.environ.get("MAX_PROXY_DOWNLOAD_BYTES", str(256 * 1024 * 1024)))
-
-
-async def _validate_public_http_target(url: str) -> str:
-    """Validate every proxy hop and reject local/private destinations."""
-    parsed = urlparse(url)
-    if parsed.scheme not in ("http", "https") or not parsed.hostname or parsed.username or parsed.password:
-        raise HTTPException(status_code=400, detail="下载地址无效")
-    try:
-        port = parsed.port
-    except ValueError:
-        raise HTTPException(status_code=400, detail="下载地址端口无效")
-    if port not in (None, 80, 443):
-        raise HTTPException(status_code=400, detail="不允许访问自定义端口")
-
-    host = parsed.hostname.rstrip(".").lower()
-    try:
-        await _resolve_public_ips(host, port or (443 if parsed.scheme == "https" else 80))
-    except PublicResolutionError as error:
-        detail = "下载地址无法解析" if error.reason == "dns_unavailable" else "下载地址解析到非公网地址"
-        raise HTTPException(status_code=400, detail=detail) from None
-    return host
-
-
-async def _send_with_validated_redirects(client, method: str, url: str, headers=None, timeout=None,
-                                         max_redirects: int = 5):
-    """逐跳校验 URL/DNS 后请求；不让 HTTPX 自动跟随未经检查的重定向。"""
-    current_url = url
-    original_host = await _validate_public_http_target(current_url)
-    original_parsed = urlparse(current_url)
-    original_origin = (
-        original_parsed.scheme.lower(),
-        original_host,
-        original_parsed.port or (443 if original_parsed.scheme.lower() == "https" else 80),
-    )
-    request_headers = dict(headers or {})
-    for redirect_count in range(max_redirects + 1):
-        current_host = await _validate_public_http_target(current_url)
-        current_parsed = urlparse(current_url)
-        current_origin = (
-            current_parsed.scheme.lower(),
-            current_host,
-            current_parsed.port or (443 if current_parsed.scheme.lower() == "https" else 80),
-        )
-        hop_headers = dict(request_headers)
-        if current_origin != original_origin:
-            # 凭据只发给初始 HTTPS origin，绝不随跨 origin 重定向传递。
-            hop_headers.pop("Authorization", None)
-        response = await client.send(
-            client.build_request(method, current_url, headers=hop_headers, timeout=timeout),
-            stream=True,
-            follow_redirects=False,
-        )
-        if response.status_code not in (301, 302, 303, 307, 308):
-            return response
-        location = response.headers.get("location")
-        await close_redirect_response(response)
-        if not location or redirect_count >= max_redirects:
-            raise HTTPException(status_code=502, detail="下载源重定向无效或次数过多")
-        current_url = urljoin(current_url, location)
-    raise HTTPException(status_code=502, detail="下载源重定向失败")
-
-
-class ProxyGithubZipRequest(BaseModel):
-    """GitHub ZIP 代理下载请求模型"""
-    url: str = Field(..., min_length=1, description="资源URL")
-    item_id: str = Field(..., min_length=1, max_length=100, description="资源ID")
-    # 🔒 P0安全加固：身份以 Authorization Token 为准，account 仅作向后兼容字段，不再作为鉴权依据
-    account: Optional[str] = Field(None, max_length=50, description="用户账号（已废弃，以Token为准）")
-    
-    @field_validator('item_id')
-    @classmethod
-    def validate_item_id(cls, v: str) -> str:
-        """验证item_id只允许[a-zA-Z0-9_-]字符"""
-        if not re.match(r'^[a-zA-Z0-9_-]+$', v):
-            raise ValueError('item_id只能包含字母、数字、下划线和连字符')
-        return v
-
-
-class SkillInstallSourceRequest(BaseModel):
-    model_config = {'extra': 'forbid'}
-    item_id: str = Field(..., min_length=1, max_length=100, pattern=r'^[A-Za-z0-9_-]+$')
-
-
-@router.post('/api/skill_install_source')
-@shared_limit(30)
-async def skill_install_source(request: Request, req_data: SkillInstallSourceRequest,
-                               account: str = Depends(require_auth), db: Session = Depends(get_db)):
-    try:
-        item = await authorized_proxy_item(json_db, db, req_data.item_id, account)
-        result = install_source(item, account, token_selector=select_github_token, signing_key=JWT_SECRET)
-    except ProxyAccessError as error:
-        return JSONResponse(content={'error': error.message}, status_code=error.status_code,
-                            headers={'Cache-Control': 'no-store'})
-    except SkillSourceError:
-        return JSONResponse(content={'error': '该资源不是可安装的Skill来源', 'error_code': 'invalid_repository'}, status_code=400,
-                            headers={'Cache-Control': 'no-store'})
-    return JSONResponse(content=result, headers={'Cache-Control': 'no-store', 'Vary': 'Authorization'})
-
-@router.post("/api/proxy_github_zip")
-@shared_limit(5)
-async def proxy_github_zip(request: Request, req_data: ProxyGithubZipRequest, account: str = Depends(require_auth), db: Session = Depends(get_db)):
-    """云端代理：校验所有权后，拉取 GitHub 私有库 ZIP 流透传给本地"""
-    try:
-        item = await authorized_proxy_item(json_db, db, req_data.item_id, account)
-    except ProxyAccessError as error:
-        return JSONResponse(content={"error": error.message}, status_code=error.status_code)
-
-    async def resolve_commit(repo_url, token):
-        # 缓存命中不加载版本引擎；保留原按需导入时序。
-        from 云端_定时版本检测引擎 import fetch_github_commit
-        return await fetch_github_commit(repo_url, token)
-
-    return await prepare_github_zip(item, req_data.item_id, store=json_db,
-        signing_key=JWT_SECRET,
-        token_selector=select_github_token, resolve_commit=resolve_commit,
-        send=_send_with_validated_redirects, client_factory=httpx.AsyncClient,
-        transport_factory=ValidatedAsyncHTTPTransport, error_response=JSONResponse,
-        cache_max_bytes=ARCHIVE_MAX_BYTES, direct_max_bytes=ZIP_DOWNLOAD_MAX_BYTES)
-
-
-
-class ProxyDownloadRequest(BaseModel):
-    """工作流/应用 JSON 代理下载请求模型"""
-    url: str = Field(..., min_length=1, description="下载URL")
-    item_id: str = Field(..., min_length=1, max_length=100, description="资源ID")
-    # 🔒 P0安全加固：身份以 Authorization Token 为准，account 仅作向后兼容字段
-    account: Optional[str] = Field(None, max_length=50, description="用户账号（已废弃，以Token为准）")
-    
-    @field_validator('item_id')
-    @classmethod
-    def validate_item_id(cls, v: str) -> str:
-        """验证item_id只允许[a-zA-Z0-9_-]字符"""
-        if not re.match(r'^[a-zA-Z0-9_-]+$', v):
-            raise ValueError('item_id只能包含字母、数字、下划线和连字符')
-        return v
-    
-    @field_validator('url')
-    @classmethod
-    def validate_url(cls, v: str) -> str:
-        """验证URL格式并阻止内网地址"""
-        if not v.startswith(('http://', 'https://')):
-            raise ValueError('url必须是有效的HTTP或HTTPS地址')
-        
-        parsed = urlparse(v)
-        # 阻止内网地址
-        dangerous_hosts = ['localhost', '127.0.0.1', '0.0.0.0', '::1']
-        if parsed.hostname in dangerous_hosts:
-            raise ValueError('不允许访问本地地址')
-        
-        # 检查是否是私有IP
-        try:
-            ip = ipaddress.ip_address(parsed.hostname)
-        except ValueError:
-            return v  # 域名仍由实际请求的逐跳DNS与传输层校验。
-        # 拒绝异常必须在解析异常处理之外，否则会被当作域名吞掉。
-        if ip.is_private or ip.is_loopback or ip.is_reserved:
-            raise ValueError('不允许访问内网地址')
-        
-        return v
-
-@router.post("/api/proxy_download")
-@shared_limit(10)
-async def proxy_download(request: Request, req_data: ProxyDownloadRequest, account: str = Depends(require_auth), db: Session = Depends(get_db)):
-    """云端代理：校验所有权后，拉取真实的工作流 JSON 文件并透传给本地"""
-    try:
-        item = await authorized_proxy_item(json_db, db, req_data.item_id, account)
-    except ProxyAccessError as error:
-        return JSONResponse(content={"error": error.message}, status_code=error.status_code)
-
-    return await prepare_workflow_download(item, req_data.item_id, store=json_db,
-        source_authorizer=authorized_workflow_path, source_error=WorkflowSourceError,
-        credential_target=workflow_credential_target, validate_target=_validate_public_http_target,
-        client_factory=httpx.AsyncClient, transport_factory=ValidatedAsyncHTTPTransport,
-        timeout_error=httpx.TimeoutException, http_error=HTTPException,
-        error_response=JSONResponse, max_bytes=MAX_PROXY_DOWNLOAD_BYTES)
+# router_proxy.py
+from fastapi import APIRouter, Depends, HTTPException, Request
+from rate_limit import shared_limit
+from fastapi.responses import StreamingResponse, JSONResponse, Response
+from pydantic import BaseModel, Field, field_validator, HttpUrl
+from typing import Optional
+from urllib.parse import urlparse, urljoin
+import ipaddress
+import httpx
+import os
+import re
+import urllib.request
+import urllib.error
+import 数据库连接 as json_db
+from database_sql import SessionLocal
+from models_sql import Ownership
+from workflow_source import authorized_workflow_path, workflow_credential_target, WorkflowSourceError
+from proxy_access import authorized_proxy_item, ProxyAccessError
+from github_source import repository_parts, select_github_token
+from github_archive import ARCHIVE_MAX_BYTES
+from 安全认证 import require_auth, JWT_SECRET
+from safe_http_transport import ValidatedAsyncHTTPTransport, _resolve_public_ips, PublicResolutionError
+from workflow_download_service import prepare_workflow_download
+from zip_stream import ZIP_DOWNLOAD_MAX_BYTES
+from github_zip_service import prepare_github_zip
+from proxy_resources import close_redirect_response
+from skill_archive_source import install_source, SkillSourceError
+
+router = APIRouter()
+MAX_PROXY_DOWNLOAD_BYTES = int(os.environ.get("MAX_PROXY_DOWNLOAD_BYTES", str(256 * 1024 * 1024)))
+
+
+async def _validate_public_http_target(url: str) -> str:
+    """Validate every proxy hop and reject local/private destinations."""
+    parsed = urlparse(url)
+    if parsed.scheme not in ("http", "https") or not parsed.hostname or parsed.username or parsed.password:
+        raise HTTPException(status_code=400, detail="下载地址无效")
+    try:
+        port = parsed.port
+    except ValueError:
+        raise HTTPException(status_code=400, detail="下载地址端口无效")
+    if port not in (None, 80, 443):
+        raise HTTPException(status_code=400, detail="不允许访问自定义端口")
+
+    host = parsed.hostname.rstrip(".").lower()
+    try:
+        await _resolve_public_ips(host, port or (443 if parsed.scheme == "https" else 80))
+    except PublicResolutionError as error:
+        detail = "下载地址无法解析" if error.reason == "dns_unavailable" else "下载地址解析到非公网地址"
+        raise HTTPException(status_code=400, detail=detail) from None
+    return host
+
+
+async def _send_with_validated_redirects(client, method: str, url: str, headers=None, timeout=None,
+                                         max_redirects: int = 5):
+    """逐跳校验 URL/DNS 后请求；不让 HTTPX 自动跟随未经检查的重定向。"""
+    current_url = url
+    original_host = await _validate_public_http_target(current_url)
+    original_parsed = urlparse(current_url)
+    original_origin = (
+        original_parsed.scheme.lower(),
+        original_host,
+        original_parsed.port or (443 if original_parsed.scheme.lower() == "https" else 80),
+    )
+    request_headers = dict(headers or {})
+    for redirect_count in range(max_redirects + 1):
+        current_host = await _validate_public_http_target(current_url)
+        current_parsed = urlparse(current_url)
+        current_origin = (
+            current_parsed.scheme.lower(),
+            current_host,
+            current_parsed.port or (443 if current_parsed.scheme.lower() == "https" else 80),
+        )
+        hop_headers = dict(request_headers)
+        if current_origin != original_origin:
+            # 凭据只发给初始 HTTPS origin，绝不随跨 origin 重定向传递。
+            hop_headers.pop("Authorization", None)
+        response = await client.send(
+            client.build_request(method, current_url, headers=hop_headers, timeout=timeout),
+            stream=True,
+            follow_redirects=False,
+        )
+        if response.status_code not in (301, 302, 303, 307, 308):
+            return response
+        location = response.headers.get("location")
+        await close_redirect_response(response)
+        if not location or redirect_count >= max_redirects:
+            raise HTTPException(status_code=502, detail="下载源重定向无效或次数过多")
+        current_url = urljoin(current_url, location)
+    raise HTTPException(status_code=502, detail="下载源重定向失败")
+
+
+class ProxyGithubZipRequest(BaseModel):
+    """GitHub ZIP 代理下载请求模型"""
+    url: str = Field(..., min_length=1, description="资源URL")
+    item_id: str = Field(..., min_length=1, max_length=100, description="资源ID")
+    # 🔒 P0安全加固：身份以 Authorization Token 为准，account 仅作向后兼容字���，不再作为鉴权依据
+    account: Optional[str] = Field(None, max_length=50, description="用户账号（已废弃，以Token为准）")
+    
+    @field_validator('item_id')
+    @classmethod
+    def validate_item_id(cls, v: str) -> str:
+        """验证item_id只允许[a-zA-Z0-9_-]字符"""
+        if not re.match(r'^[a-zA-Z0-9_-]+$', v):
+            raise ValueError('item_id只能包含字母、数字、下划线和连字符')
+        return v
+
+
+class SkillInstallSourceRequest(BaseModel):
+    model_config = {'extra': 'forbid'}
+    item_id: str = Field(..., min_length=1, max_length=100, pattern=r'^[A-Za-z0-9_-]+$')
+
+
+@router.post('/api/skill_install_source')
+@shared_limit(30)
+async def skill_install_source(request: Request, req_data: SkillInstallSourceRequest,
+                               account: str = Depends(require_auth)):
+    try:
+        item = await authorized_proxy_item(json_db, SessionLocal, req_data.item_id, account)
+        result = install_source(item, account, token_selector=select_github_token, signing_key=JWT_SECRET)
+    except ProxyAccessError as error:
+        return JSONResponse(content={'error': error.message}, status_code=error.status_code,
+                            headers={'Cache-Control': 'no-store'})
+    except SkillSourceError:
+        return JSONResponse(content={'error': '该资源不是可安装的Skill来源', 'error_code': 'invalid_repository'}, status_code=400,
+                            headers={'Cache-Control': 'no-store'})
+    return JSONResponse(content=result, headers={'Cache-Control': 'no-store', 'Vary': 'Authorization'})
+
+@router.post("/api/proxy_github_zip")
+@shared_limit(5)
+async def proxy_github_zip(request: Request, req_data: ProxyGithubZipRequest, account: str = Depends(require_auth)):
+    """云端代理：校验所有权后，拉取 GitHub 私有库 ZIP 流透传给本地"""
+    try:
+        item = await authorized_proxy_item(json_db, SessionLocal, req_data.item_id, account)
+    except ProxyAccessError as error:
+        return JSONResponse(content={"error": error.message}, status_code=error.status_code)
+
+    async def resolve_commit(repo_url, token):
+        # 缓存命中不加载版本引擎；保留原按需导入时序。
+        from 云端_定时版本检测引擎 import fetch_github_commit
+        return await fetch_github_commit(repo_url, token)
+
+    return await prepare_github_zip(item, req_data.item_id, store=json_db,
+        signing_key=JWT_SECRET,
+        token_selector=select_github_token, resolve_commit=resolve_commit,
+        send=_send_with_validated_redirects, client_factory=httpx.AsyncClient,
+        transport_factory=ValidatedAsyncHTTPTransport, error_response=JSONResponse,
+        cache_max_bytes=ARCHIVE_MAX_BYTES, direct_max_bytes=ZIP_DOWNLOAD_MAX_BYTES)
+
+
+
+class ProxyDownloadRequest(BaseModel):
+    """工作流/应用 JSON 代理下载请求模型"""
+    url: str = Field(..., min_length=1, description="下载URL")
+    item_id: str = Field(..., min_length=1, max_length=100, description="资源ID")
+    # 🔒 P0安全加固：身份以 Authorization Token 为准，account 仅作向后兼容字段
+    account: Optional[str] = Field(None, max_length=50, description="用户账号（已废弃，以Token为准）")
+    
+    @field_validator('item_id')
+    @classmethod
+    def validate_item_id(cls, v: str) -> str:
+        """验证item_id只允许[a-zA-Z0-9_-]字符"""
+        if not re.match(r'^[a-zA-Z0-9_-]+$', v):
+            raise ValueError('item_id只能包含字母、数字、下划线和连字符')
+        return v
+    
+    @field_validator('url')
+    @classmethod
+    def validate_url(cls, v: str) -> str:
+        """验证URL格式并阻止内网地址"""
+        if not v.startswith(('http://', 'https://')):
+            raise ValueError('url必须是有效的HTTP或HTTPS地址')
+        
+        parsed = urlparse(v)
+        # 阻止内网地址
+        dangerous_hosts = ['localhost', '127.0.0.1', '0.0.0.0', '::1']
+        if parsed.hostname in dangerous_hosts:
+            raise ValueError('不允许访问本地地址')
+        
+        # 检查是否是私有IP
+        try:
+            ip = ipaddress.ip_address(parsed.hostname)
+        except ValueError:
+            return v  # 域名仍由实际请求的逐跳DNS与传输层校验。
+        # 拒绝异常必须在解析异常处理之外，否则会被当作域名吞掉。
+        if ip.is_private or ip.is_loopback or ip.is_reserved:
+            raise ValueError('不允许访问内网地址')
+        
+        return v
+
+@router.post("/api/proxy_download")
+@shared_limit(10)
+async def proxy_download(request: Request, req_data: ProxyDownloadRequest, account: str = Depends(require_auth)):
+    """云端代理：校验所有权后，拉取真实的工作流 JSON 文件并透传给本地"""
+    try:
+        item = await authorized_proxy_item(json_db, SessionLocal, req_data.item_id, account)
+    except ProxyAccessError as error:
+        return JSONResponse(content={"error": error.message}, status_code=error.status_code)
+
+    return await prepare_workflow_download(item, req_data.item_id, store=json_db,
+        source_authorizer=authorized_workflow_path, source_error=WorkflowSourceError,
+        credential_target=workflow_credential_target, validate_target=_validate_public_http_target,
+        client_factory=httpx.AsyncClient, transport_factory=ValidatedAsyncHTTPTransport,
+        timeout_error=httpx.TimeoutException, http_error=HTTPException,
+        error_response=JSONResponse, max_bytes=MAX_PROXY_DOWNLOAD_BYTES)