fix(security): harden public runtime boundaries

Author: AgentArcLab

ksadk/builders/container_builder.py

@@ -199,15 +199,22 @@ def _optimize_kcr_endpoint(self, registry: str) -> str:
         """
         from ksadk.configs.settings import check_endpoint_reachable
 
+        registry_text = str(registry).strip()
+        scheme, separator, reference = registry_text.partition("://")
+        prefix = f"{scheme}://" if separator else ""
+        target = reference if separator else registry_text
+        host, slash, remainder = target.partition("/")
+        normalized_host = host.lower()
+
         # 已经是内网地址
-        if "hub-vpc" in registry:
+        if normalized_host == "hub-vpc.kce.ksyun.com":
             return registry
 
-        # 匹配企业版 KCR: hub.kce.ksyun.com
-        if "hub.kce.ksyun.com" in registry:
+        # 仅匹配完整 registry host，避免改写路径或攻击者域名中的同名子串。
+        if normalized_host == "hub.kce.ksyun.com":
             click.echo("🔍 检测 KCR 内网连通性...")
             if check_endpoint_reachable("hub-vpc.kce.ksyun.com", port=443, timeout=2.0):
-                optimized = registry.replace("hub.kce.ksyun.com", "hub-vpc.kce.ksyun.com")
+                optimized = f"{prefix}hub-vpc.kce.ksyun.com{slash}{remainder}"
                 click.secho(f"   ✅ 优化为内网: {optimized}", fg="green")
                 return optimized
             else:
@@ -522,7 +529,7 @@ def _generate_entrypoint(self, detection_result, package_name: str) -> str:
     entry_point="{detection_result.entry_point}",
     package_path="/app/{package_name}",
     agent_variable="{detection_result.agent_variable}",
-    runner_class="{getattr(detection_result, 'runner_class', '')}"
+    runner_class="{getattr(detection_result, "runner_class", "")}"
 )
 
 logger.info(f"框架: {{detection_result.name}}")

ksadk/cli/cmd_create.py

@@ -624,17 +624,17 @@ def _fix_nested_imports(file_path: Path, subdirs: list[str]) -> bool:
 
 
 def _generate_env_content(global_env: dict) -> str:
-    """生成 .env 文件内容"""
-    api_key = global_env.get("OPENAI_API_KEY", "")
+    """Generate project-local configuration without copying credentials.
+
+    ``init`` may read a global profile to retain non-secret endpoint defaults, but
+    it must not persist the caller's credentials in a newly created directory.
+    """
     base_url = global_env.get("OPENAI_BASE_URL", "")
     model_name = global_env.get("OPENAI_MODEL_NAME", "")
-    ks_ak = global_env.get("KSYUN_ACCESS_KEY", "")
-    ks_sk = global_env.get("KSYUN_SECRET_KEY", "")
     ks_region = global_env.get("KSYUN_REGION", "cn-beijing-6")
-    ks_account = global_env.get("KSYUN_ACCOUNT_ID", "")
 
-    env_content = f"""# 模型配置
-OPENAI_API_KEY={api_key}
+    env_content = """# 模型配置
+# OPENAI_API_KEY=
 """
     if base_url:
         env_content += f"OPENAI_BASE_URL={base_url}\n"
@@ -648,19 +648,10 @@ def _generate_env_content(global_env: dict) -> str:
     env_content += """
 # 金山云配置
 """
-    if ks_ak:
-        env_content += f"KSYUN_ACCESS_KEY={ks_ak}\n"
-    else:
-        env_content += "# KSYUN_ACCESS_KEY=\n"
-    if ks_sk:
-        env_content += f"KSYUN_SECRET_KEY={ks_sk}\n"
-    else:
-        env_content += "# KSYUN_SECRET_KEY=\n"
+    env_content += "# KSYUN_ACCESS_KEY=\n"
+    env_content += "# KSYUN_SECRET_KEY=\n"
     env_content += f"KSYUN_REGION={ks_region}\n"
-    if ks_account:
-        env_content += f"KSYUN_ACCOUNT_ID={ks_account}\n"
-    else:
-        env_content += "# KSYUN_ACCOUNT_ID=\n"
+    env_content += "# KSYUN_ACCOUNT_ID=\n"
 
     return env_content
 
@@ -674,12 +665,11 @@ def _generate_codex_env_content(global_env: dict) -> str:
     optional proxy knobs are documented rather than copied from the global profile,
     because their upstream key can be different from the model key.
     """
-    api_key = global_env.get("OPENAI_API_KEY", "")
     base_url = global_env.get("OPENAI_BASE_URL", "")
     model_name = global_env.get("OPENAI_MODEL_NAME", "")
     lines = [
         "# 本地 Codex 模型配置（仅用于 ksadk web，不会进入 ManagedRuntime bundle）",
-        f"OPENAI_API_KEY={api_key}",
+        "# OPENAI_API_KEY=",
         f"OPENAI_BASE_URL={base_url}" if base_url else "# OPENAI_BASE_URL=",
         f"OPENAI_MODEL_NAME={model_name}" if model_name else "# OPENAI_MODEL_NAME=glm-5.2",
         "",
@@ -1208,7 +1198,7 @@ def _wrap_agent_file(from_agent_path: Path, project_name: str, framework: str, a
     if global_config_exists():
         global_env = get_env_from_global_config()
         if global_env:
-            print_info("检测到全局配置，已自动填充凭证")
+            print_info("检测到全局配置；项目 .env 仅保留非敏感默认值，凭证不会被复制")
 
     # 生成 .env
     (project_path / ".env").write_text(_generate_env_content(global_env), encoding="utf-8-sig")
@@ -1385,7 +1375,7 @@ def _ignore_copytree(_dir: str, names: list[str]):
     if global_config_exists():
         global_env = get_env_from_global_config()
         if global_env:
-            print_info("检测到全局配置，已自动填充凭证")
+            print_info("检测到全局配置；项目 .env 仅保留非敏感默认值，凭证不会被复制")
 
     # 生成 .env
     (project_path / ".env").write_text(_generate_env_content(global_env), encoding="utf-8-sig")
@@ -1808,35 +1798,27 @@ def create(project_name: str, framework: str, from_agent_path: str):
     if global_config_exists():
         global_env = get_env_from_global_config()
         if global_env:
-            print_info("检测到全局配置，已自动填充凭证")
+            print_info("检测到全局配置；项目 .env 仅保留非敏感默认值，凭证不会被复制")
 
     # .env - 生成配置文件
-    # 如果有全局配置，使用全局配置的值；否则使用占位符
-    # 如果有全局配置，使用全局配置的值；否则使用空字符串
-    api_key = global_env.get("OPENAI_API_KEY", "")
+    # 仅继承非敏感的 endpoint/model/region 默认值；绝不复制凭证。
     base_url = global_env.get("OPENAI_BASE_URL", "")
     model_name = global_env.get("OPENAI_MODEL_NAME", "")
 
-    ks_ak = global_env.get("KSYUN_ACCESS_KEY", "")
-    ks_sk = global_env.get("KSYUN_SECRET_KEY", "")
     ks_region = global_env.get("KSYUN_REGION", "cn-beijing-6")
-    ks_account = global_env.get("KSYUN_ACCOUNT_ID", "")
 
     # 构建 .env 内容
     if framework == "openclaw":
         env_content = f"""# ======================
 # OpenClaw 标准部署最小配置
 # ======================
-KSYUN_ACCESS_KEY={ks_ak}
-KSYUN_SECRET_KEY={ks_sk}
+# KSYUN_ACCESS_KEY=
+# KSYUN_SECRET_KEY=
 KSYUN_REGION={ks_region}
 """
-        if ks_account:
-            env_content += f"KSYUN_ACCOUNT_ID={ks_account}\n"
-        else:
-            env_content += "# KSYUN_ACCOUNT_ID=your-account-id\n"
+        env_content += "# KSYUN_ACCOUNT_ID=your-account-id\n"
 
-        env_content += f"\nOPENAI_API_KEY={api_key}\n"
+        env_content += "\n# OPENAI_API_KEY=\n"
         if base_url:
             env_content += f"OPENAI_BASE_URL={base_url}\n"
         else:
@@ -1850,18 +1832,13 @@ def create(project_name: str, framework: str, from_agent_path: str):
         env_content = f"""# ======================
 # Hermes 标准部署最小配置
 # ======================
-KSYUN_ACCESS_KEY={ks_ak}
-KSYUN_SECRET_KEY={ks_sk}
+# KSYUN_ACCESS_KEY=
+# KSYUN_SECRET_KEY=
 KSYUN_REGION={ks_region}
 """
-        if ks_account:
-            env_content += f"KSYUN_ACCOUNT_ID={ks_account}\n"
-        else:
-            env_content += "# KSYUN_ACCOUNT_ID=your-account-id\n"
+        env_content += "# KSYUN_ACCOUNT_ID=your-account-id\n"
 
-        env_content += f"""
-OPENAI_API_KEY={api_key}
-"""
+        env_content += "\n# OPENAI_API_KEY=\n"
         if base_url:
             env_content += f"OPENAI_BASE_URL={base_url}\n"
         else:
@@ -1908,14 +1885,12 @@ def create(project_name: str, framework: str, from_agent_path: str):
     elif framework == "codex":
         env_content = _generate_codex_env_content(global_env)
     else:
-        langfuse_public = global_env.get("LANGFUSE_PUBLIC_KEY", "")
-        langfuse_secret = global_env.get("LANGFUSE_SECRET_KEY", "")
         langfuse_url = global_env.get("LANGFUSE_BASE_URL", "")
 
-        env_content = f"""# ======================
+        env_content = """# ======================
 # 模型配置 (必填, 可以从星流平台获取https://ksp.console.ksyun.com/#/apiKey)
 # ======================
-OPENAI_API_KEY={api_key}
+# OPENAI_API_KEY=
 """
 
         # 可选字段：如果有值则启用，否则注释掉
@@ -1934,15 +1909,8 @@ def create(project_name: str, framework: str, from_agent_path: str):
 # 可观测性 (可选)
 # ======================
 """
-        if langfuse_public:
-            env_content += f"LANGFUSE_PUBLIC_KEY={langfuse_public}\n"
-        else:
-            env_content += "# LANGFUSE_PUBLIC_KEY=pk-xxx\n"
-
-        if langfuse_secret:
-            env_content += f"LANGFUSE_SECRET_KEY={langfuse_secret}\n"
-        else:
-            env_content += "# LANGFUSE_SECRET_KEY=sk-xxx\n"
+        env_content += "# LANGFUSE_PUBLIC_KEY=pk-xxx\n"
+        env_content += "# LANGFUSE_SECRET_KEY=sk-xxx\n"
 
         if langfuse_url:
             env_content += f"LANGFUSE_BASE_URL={langfuse_url}\n"
@@ -1954,22 +1922,12 @@ def create(project_name: str, framework: str, from_agent_path: str):
 # 金山云配置 (可选,需要部署时必选)
 # ======================
 """
-        if ks_ak:
-            env_content += f"KSYUN_ACCESS_KEY={ks_ak}\n"
-        else:
-            env_content += "# KSYUN_ACCESS_KEY=your-api-key-here\n"
-
-        if ks_sk:
-            env_content += f"KSYUN_SECRET_KEY={ks_sk}\n"
-        else:
-            env_content += "# KSYUN_SECRET_KEY=your-api-secret-here\n"
+        env_content += "# KSYUN_ACCESS_KEY=your-api-key-here\n"
+        env_content += "# KSYUN_SECRET_KEY=your-api-secret-here\n"
 
         env_content += f"KSYUN_REGION={ks_region}\n"
 
-        if ks_account:
-            env_content += f"KSYUN_ACCOUNT_ID={ks_account}\n"
-        else:
-            env_content += "# KSYUN_ACCOUNT_ID=your-account-id\n"
+        env_content += "# KSYUN_ACCOUNT_ID=your-account-id\n"
 
     # 使用 utf-8-sig 编码 (带 BOM)，确保 Windows 程序正确识别为 UTF-8
     (project_path / ".env").write_text(env_content, encoding="utf-8-sig")

ksadk/model_proxy/cache.py

@@ -5,7 +5,7 @@
 "模型不存在"而非"endpoint 不存在"。本模块:
 
 - 键 = ``(model, base_url, credential_scope)``;credential_scope 用 key 的稳定指纹
-  (sha256 前 8 位),避免明文 key 入键/日志,又能区分不同 key。
+  (BLAKE2b 前 8 字节),避免明文 key 入键/日志,又能区分不同 key。
 - **明确判定**(supported/unsupported)长缓存(TTL);**不确定**(unknown)不缓存,
   下次重探——避免把一次抖动固化成长期误判。
 - **singleflight**:并发同键只探一次,其余等结果(用 per-key Future)。
@@ -26,8 +26,10 @@
 
 
 def _scope(key: str) -> str:
-    """key 的稳定指纹(sha256 前 8 位),避免明文入键/日志。"""
-    return hashlib.sha256(key.encode("utf-8")).hexdigest()[:8]
+    """Return a domain-separated, non-secret cache scope for a credential."""
+    return hashlib.blake2b(
+        key.encode("utf-8"), digest_size=8, person=b"ksadk-capability"
+    ).hexdigest()
 
 
 @dataclass

ksadk/model_proxy/demo_webui.py

@@ -13,12 +13,15 @@
 from __future__ import annotations
 
 import json
+import logging
 import os
 
 import uvicorn
 from fastapi import FastAPI, Request
 from fastapi.responses import HTMLResponse, StreamingResponse
 
+logger = logging.getLogger(__name__)
+
 # ruff: noqa: E501  内联 HTML/CSS/JS 演示字面量,长行不拆
 
 _PAGE = """<!doctype html><html lang="zh"><head><meta charset="utf-8">
@@ -83,9 +86,11 @@ async def gen():
                         if it.get("type") == "agentMessage" and it.get("text"):
                             yield f"data: {json.dumps({'reply': it['text']}, ensure_ascii=False)}\n\n"
                     elif m == "error":
-                        yield f"data: {json.dumps({'error': json.dumps(ev.get('params', {}).get('error', {}), ensure_ascii=False)[:200]}, ensure_ascii=False)}\n\n"
+                        logger.warning("Codex demo runtime reported an error")
+                        yield 'data: {"error":"Codex runtime failed; see local logs."}\n\n'
             except Exception as e:  # noqa: BLE001
-                yield f"data: {json.dumps({'error': str(e)}, ensure_ascii=False)}\n\n"
+                logger.warning("Codex demo runtime failed", exc_info=e)
+                yield 'data: {"error":"Codex runtime failed; see local logs."}\n\n'
 
         return StreamingResponse(gen(), media_type="text/event-stream")