Back to skill

Security audit

TeamClaw

Security checks for vulnerabilities and agentic risk

Overview

TeamClaw is a disclosed multi-agent service, but it grants broad local execution, credential, session, and network authority with several under-scoped security controls.

Install only if you intend to run a broad local agent platform and can isolate it. Keep it on loopback unless you add strong authentication, avoid the public tunnel on untrusted networks, do not expose OASIS directly, rotate any configured OpenClaw/API/internal tokens after testing, and treat the command/code execution tools as full local code execution rather than a sandbox.

Vulnerability Patterns
  • Agent Memory PoisoningWrites attacker-controlled rules into memory that affect later sessions
  • Remote Payload Retrieval and ExecutionFetches external code whose behavior can change after review
  • Unauthorized Access and Privilege EscalationObtains permissions beyond the task's legitimate needs
  • Insecure DependenciesIntroduces malicious components through unsafe dependency sources
  • Insecure Skill Coding PracticesFinds exploitable flaws such as hardcoded secrets or command injection
Findings (7)

T05 · Unauthorized Access and Privilege Escalation

Error
Location
oasis/server.py:122
Finding
Administrator Token Exfiltration Through an Attacker-Controlled Callback<![CDATA[ ## Vulnerability Details **File Location**: `oasis/server.py:122-143`, `oasis/server.py:153-179` **Vulnerability Type**: Server-Side Request Forgery and credential disclosure **Risk Level**: Critical ### Vulnerable Code ```python cb_url = getattr(engine, "callback_url", None) if cb_url: conclusion = forum.conclusion if forum else "(无结论)" status = forum.status if forum else "error" cb_session = getattr(engine, "callback_session_id", "default") or "default" user_id = forum.user_id if forum else "anonymous" internal_token = os.getenv("INTERNAL_TOKEN", "") text = ( f"[OASIS 子任务完成通知]\n" f"Topic ID: {topic_id}\n" f"状态: {status}\n" f"主题: {forum.question if forum else '?'}\n\n" f"📋 结论:\n{conclusion}" ) try: async with httpx.AsyncClient(timeout=10.0) as client: await client.post( cb_url, json={"user_id": user_id, "text": text, "session_id": cb_session}, headers={"X-Internal-Token": internal_token}, ) ``` ```python @app.post("/topics", response_model=dict) async def create_topic(req: CreateTopicRequest): """Create a new discussion topic. Returns topic_id for tracking.""" topic_id = str(uuid.uuid4())[:8] forum = DiscussionForum( topic_id=topic_id, question=req.question, user_id=req.user_id, max_rounds=req.max_rounds, ) discussions[topic_id] = forum forum.save() engine = DiscussionEngine( forum=forum, schedule_yaml=req.schedule_yaml, schedule_file=req.schedule_file, bot_enabled_tools=req.bot_enabled_tools, bot_timeout=req.bot_timeout, user_id=req.user_id, early_stop=req.early_stop, discussion=req.discussion, ) engine.callback_url = req.callback_url engine.callback_session_id = req.callback_session_id ``` ### Technical Analysis The unauthenticated `/topics` endpoint accepts ...[truncated 1874 chars]
Remediation
<![CDATA[ ## Remediation Suggestions - Require authentication on `/topics` and every other OASIS route. - Derive `user_id` from the authenticated principal instead of accepting it as an authority-bearing request field. - Do not send `INTERNAL_TOKEN` to a URL supplied by a client. - Replace arbitrary callbacks with preconfigured callback identifiers mapped to exact trusted destinations. - If callbacks must be configurable, enforce an exact scheme, hostname, port, and path allowlist. - Resolve hostnames and reject loopback, link-local, private, multicast, metadata, and reserved addresses unless explicitly required. - Disable HTTP redirects or revalidate every redirect destination. - Use narrowly scoped, short-lived, audience-bound callback credentials instead of the global internal token. - Rotate `INTERNAL_TOKEN` after remediation because existing deployments may already have exposed it. ]]>

T05 · Unauthorized Access and Privilege Escalation

Error
Location
oasis/server.py:659
Finding
Unauthenticated OASIS Administration and OpenClaw API-Key Disclosure<![CDATA[ ## Vulnerability Details **File Location**: `oasis/server.py:333-374`, `oasis/server.py:522-640`, `oasis/server.py:659-714`; documented in `SKILL.md:337-347` **Vulnerability Type**: Missing authentication and sensitive configuration disclosure **Risk Level**: Critical ### Vulnerable Code ```python @app.get("/sessions/oasis") async def list_oasis_sessions(user_id: str = Query("")): """List all oasis-managed sessions by scanning the agent checkpoint DB. Query param: user_id (optional). If provided, only sessions for that user are returned. """ db_path = os.path.join(_project_root, "data", "agent_memory.db") ``` ```python @app.post("/workflows") async def save_workflow(req: WorkflowSaveRequest): """Save a YAML workflow under data/user_files/{user}/oasis/yaml/.""" user = req.user_id name = req.name ``` ```python @app.get("/sessions/openclaw") async def list_openclaw_sessions(filter: str = Query("")): """List OpenClaw sessions from sessions.json file.""" if not os.path.exists(_OPENCLAW_SESSIONS_FILE): return {"sessions": [], "available": False, "message": "OpenClaw sessions file not found"} try: with open(_OPENCLAW_SESSIONS_FILE, "r", encoding="utf-8") as f: data = json.load(f) except Exception as e: raise HTTPException(500, f"Failed to read OpenClaw sessions: {e}") # Support both dict (key->session) and list formats sessions = [] if isinstance(data, dict): for k, v in data.items(): if isinstance(v, dict): v.setdefault("key", k) sessions.append(v) elif isinstance(data, list): sessions = data # Keyword filter if filter: sessions = [s for s in sessions if filter.lower() in s.get("key", "").lower()] # Sort by updatedAt descending sessions.sort(key=lambda s: s.get("updatedAt", 0), reverse=True) result = [ { "key": s.get("key"), "sessionId" ...[truncated 2277 chars]
Remediation
<![CDATA[ ## Remediation Suggestions - Add centralized authentication middleware to every OASIS route. - Use authenticated identities for authorization and tenant selection. - Do not accept `user_id` as proof of ownership. - Add explicit per-resource authorization checks for reads, writes, updates, and deletion. - Remove `openclaw_api_key` from all API responses. - Keep downstream secrets only in server-side secret storage. - Return only minimal, non-sensitive OpenClaw metadata. - Add rate limiting and quotas to discussion-creation endpoints. - Restrict OASIS to loopback and apply host firewall rules as defense in depth. - Add tests proving that one user cannot enumerate or modify another user's resources. ]]>

T09 · Insecure Skill Coding Practices

Error
Location
src/mcp_commander.py:194
Finding
Command Sandbox Permits Unrestricted Native Code Execution<![CDATA[ ## Vulnerability Details **File Location**: `src/mcp_commander.py:36-76`, `src/mcp_commander.py:194-218`, `src/mcp_commander.py:258-280`; default enablement in `src/agent.py:91-102`, `src/agent.py:393-401` **Vulnerability Type**: Ineffective sandbox and arbitrary code execution **Risk Level**: Critical ### Vulnerable Code ```python _DEFAULT_COMMANDS = { # File and directory utilities "ls", "cat", "head", "tail", "wc", "du", "find", "file", "stat", # Text processing "grep", "awk", "sed", "sort", "uniq", "cut", "tr", "diff", "comm", # System information "echo", "date", "cal", "whoami", "uname", "hostname", "uptime", "free", "df", "env", "printenv", # Utilities "pwd", "which", "expr", "seq", "yes", "true", "false", "base64", "md5sum", "sha256sum", "xxd", # Python "python", "python3", # Network "ping", "curl", "wget", "npm", "npx", "git", "node" } ``` ```python proc = await asyncio.create_subprocess_shell( command, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, cwd=workspace, env=_sandbox_env(workspace, username), ) ``` ```python @mcp.tool() async def run_python_code(username: str, code: str) -> str: workspace = _user_workspace(username) tmp_script = os.path.join(workspace, ".tmp_exec.py") try: with open(tmp_script, "w", encoding="utf-8") as f: f.write(code) proc = await asyncio.create_subprocess_exec( _python_cmd(), tmp_script, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, cwd=workspace, env=_sandbox_env(workspace, username), ) ``` ```python enabled_names = state.get("enabled_tools") if enabled_names is not None: enabled_set = set(enabled_names) else: enabled_set = None # None = all allowed ``` ```python if enabled_names is not None: filtered_tools = [t for t in all_tools if t.name in enabled_names] else: filter ...[truncated 2023 chars]
Remediation
<![CDATA[ ## Remediation Suggestions - Remove `run_python_code` from the default tool set. - Default to no execution tools unless a trusted administrator explicitly enables them. - Remove interpreters, package managers, downloaders, and source-control clients from command allowlists. - Do not use `shell=True` or `create_subprocess_shell` for untrusted input. - Implement supported operations as fixed executable paths with separately validated arguments. - Run execution tasks in a disposable container or microVM with: - A dedicated unprivileged UID. - A read-only root filesystem. - A narrowly mounted per-user workspace. - No access to configuration files or host sockets. - Network disabled by default. - CPU, memory, process, file-size, and time limits. - Seccomp, AppArmor, SELinux, or equivalent controls. - Require explicit user confirmation that displays the exact command or code before execution. - Treat all model-generated tool arguments as untrusted input. ]]>

T03 · Remote Payload Retrieval and Execution

Error
Location
scripts/setup_env.sh:13
Finding
Mutable Remote Scripts and Binaries Are Downloaded and Executed Without Integrity Verification<![CDATA[ ## Vulnerability Details **File Location**: `scripts/setup_env.sh:13-18`, `scripts/launcher.py:70-99`, `scripts/launcher.py:196-225`; invoked by `selfskill/scripts/run.sh:119-122` **Vulnerability Type**: Unverified remote payload retrieval and execution **Risk Level**: High ### Vulnerable Code ```bash if command -v uv &>/dev/null; then echo "✅ uv is installed: $(uv --version)" else echo "Installing uv..." curl -LsSf https://astral.sh/uv/install.sh | sh export PATH="$HOME/.local/bin:$PATH" ``` ```python def download_bark_server(): """Download bark-server binary to bin/ directory.""" os_name, arch = detect_bark_platform() if not os_name: return False url = f"https://github.com/Finb/bark-server/releases/latest/download/bark-server_{os_name}_{arch}" bin_dir = os.path.join(PROJECT_ROOT, "bin") os.makedirs(bin_dir, exist_ok=True) try: urllib.request.urlretrieve(url, BARK_SERVER_PATH) os.chmod( BARK_SERVER_PATH, os.stat(BARK_SERVER_PATH).st_mode | stat.S_IXUSR | stat.S_IXGRP | stat.S_IXOTH, ) return True ``` ```python bark_proc = subprocess.Popen( [BARK_SERVER_PATH, "-addr", f"127.0.0.1:{PORT_BARK}", "-data", bark_data_dir], cwd=PROJECT_ROOT, stdout=None, stderr=None, ) ``` ```bash setup) echo "=== Environment setup ===" bash scripts/setup_env.sh echo "=== Environment setup complete ===" ;; ``` ### Technical Analysis The documented setup path pipes a remote HTTP response directly into a shell. The payload can change after the Skill has been reviewed. Normal startup separately downloads a Bark executable from a mutable `latest` release URL, marks it executable, and launches it. The code does not pin an immutable version or verify a checksum, signature, expected size, or trusted release manifest. HTTPS protects transport under normal conditions but does not ensure that the publisher account, mutable release, bui ...[truncated 1031 chars]
Remediation
<![CDATA[ ## Remediation Suggestions - Never pipe downloaded content directly into a shell. - Pin the uv installer and Bark server to immutable, reviewed versions. - Download artifacts to temporary files before processing. - Verify vendor-published SHA-256 or stronger hashes. - Verify cryptographic release signatures against pinned trusted keys. - Abort installation if verification fails. - Avoid `releases/latest`; use an immutable release tag and artifact digest. - Prefer distribution packages, vendored binaries, or reproducible local builds. - Store verified artifacts in a controlled internal repository. - Document the exact versions and verification procedure. ]]>

T09 · Insecure Skill Coding Practices

Error
Location
oasis/server.py:522
Finding
Unauthenticated Path Traversal in Workflow and Discussion Persistence<![CDATA[ ## Vulnerability Details **File Location**: `oasis/server.py:522-544`, `oasis/server.py:573-591`, `oasis/forum.py:139-149` **Vulnerability Type**: Path traversal and unauthorized file access **Risk Level**: High ### Vulnerable Code ```python @app.post("/workflows") async def save_workflow(req: WorkflowSaveRequest): """Save a YAML workflow under data/user_files/{user}/oasis/yaml/.""" user = req.user_id name = req.name if not name.endswith((".yaml", ".yml")): name += ".yaml" try: data = _yaml.safe_load(req.schedule_yaml) if not isinstance(data, dict) or "plan" not in data: raise ValueError("must contain 'plan'") except Exception as e: raise HTTPException(400, f"YAML parse failed: {e}") yaml_dir = os.path.join(_project_root, "data", "user_files", user, "oasis", "yaml") os.makedirs(yaml_dir, exist_ok=True) filepath = os.path.join(yaml_dir, name) content = (f"# {req.description}\n" if req.description else "") + req.schedule_yaml try: with open(filepath, "w", encoding="utf-8") as f: f.write(content) ``` ```python @app.get("/workflows") async def list_workflows(user_id: str = Query(...)): yaml_dir = os.path.join(_project_root, "data", "user_files", user_id, "oasis", "yaml") if not os.path.isdir(yaml_dir): return {"workflows": []} ``` ```python def _storage_path(self) -> str: user_dir = os.path.join(DISCUSSIONS_DIR, self.user_id) os.makedirs(user_dir, exist_ok=True) return os.path.join(user_dir, f"{self.topic_id}.json") def save(self): """Persist current state to disk.""" path = self._storage_path() with open(path, "w", encoding="utf-8") as f: json.dump(self.to_dict(), f, ensure_ascii=False, indent=2) ``` ### Technical Analysis The OASIS server joins caller-controlled `user_id` and workflow `name` values directly into filesystem paths. It does not reject absolute paths or traversal components, reso ...[truncated 1438 chars]
Remediation
<![CDATA[ ## Remediation Suggestions - Derive the tenant directory from the authenticated principal. - Restrict user and workflow identifiers to a conservative allowlist such as letters, digits, underscores, and hyphens. - Reject absolute paths, path separators, null bytes, and `..` components. - Resolve both the storage root and candidate path with `Path.resolve()`. - Require `candidate.relative_to(storage_root)` to succeed before every read, write, list, or delete. - Reject symlink components or use descriptor-relative filesystem operations with no-follow semantics. - Generate server-side filenames rather than trusting client-provided paths. - Apply identical validation to workflow, layout, topic, expert, and session persistence. - Add regression tests for absolute paths, traversal, mixed separators, encoded traversal, and symlink escapes. ]]>

T02 · Agent Memory Poisoning

Error
Location
src/agent.py:449
Finding
Persistent User Profile Is Injected Into the System Prompt Without Trust Separation<![CDATA[ ## Vulnerability Details **File Location**: `data/prompts/base_system.txt:59-69`, `src/agent.py:211-219`, `src/agent.py:449-465` **Vulnerability Type**: Persistent prompt and memory poisoning **Risk Level**: High ### Vulnerable Code ```python def _get_user_profile(self, user_id: str) -> str: """Read user profile from data/user_files/{user_id}/user_profile.txt.""" user_files_dir = self._prompts.get("_user_files_dir", "") fpath = os.path.join(user_files_dir, user_id, "user_profile.txt") try: with open(fpath, "r", encoding="utf-8") as f: return f.read().strip() except FileNotFoundError: return "" ``` ```python base_prompt = ( self._prompts["base_system"] + "\n\n" f"【Default available tools】\n{all_tool_list_str}\n" "All tools above are enabled by default.\n" ) if not is_subagent: user_profile = self._get_user_profile(user_id) if user_profile: base_prompt += f"\n{user_profile}\n" base_prompt += self._get_user_skills(user_id) + "\n" ``` ```python input_messages = [SystemMessage(content=base_prompt)] + history_messages ``` The base prompt also instructs the agent to proactively update `user_profile.txt` when it infers valuable user characteristics. ### Technical Analysis Profile data can be influenced by user conversation and written by the agent. On later requests, the profile file is concatenated directly into `base_prompt`, which is then sent as a privileged `SystemMessage`. There is no structured schema, provenance annotation, quoting, escaping, instruction filtering, or separation between trusted policy and untrusted remembered data. Consequently, instruction-like text stored in the profile receives system-message placement in future sessions. This changes a transient prompt-injection attempt into a persistent influence channel. The risk is amplified because all tools, including code execution, are enabled when no explicit tool list is supplied. ### Attack Path 1 ...[truncated 1131 chars]
Remediation
<![CDATA[ ## Remediation Suggestions - Do not concatenate memory content into the trusted system prompt. - Represent memory as a separate, explicitly untrusted data message. - Store only schema-validated factual fields rather than arbitrary free-form text. - Reject imperative language and instruction-like content from profile fields. - Require explicit user confirmation before adding or changing persistent profile data. - Record provenance, timestamps, and the source message for each profile entry. - Provide users with review, correction, export, and deletion controls. - Limit which tools can be influenced by remembered data. - Add a fixed system instruction stating that memory is reference data and must never override policy or current user intent. ]]>

T08 · Insecure Dependencies

Warning
Location
config/requirements.txt:1
Finding
All Python Dependencies Are Installed Without Version or Hash Pinning<![CDATA[ ## Vulnerability Details **File Location**: `config/requirements.txt:1-30`, `scripts/setup_env.sh:39-41` **Vulnerability Type**: Non-reproducible and mutable dependency installation **Risk Level**: Medium ### Vulnerable Code ```text flask requests fastapi uvicorn pydantic langgraph langchain-openai langchain-google-genai langchain-anthropic langchain-deepseek langchain-core langchain-mcp-adapters python-dotenv httpx mcp apscheduler aiosqlite langgraph-checkpoint-sqlite ddgs pymupdf qq-botpy python-telegram-bot aiohttp aiohttp_socks silk-python pydub av static_ffmpeg ``` ```bash echo "Installing dependencies from config/requirements.txt..." uv pip install -r config/requirements.txt ``` ### Technical Analysis The dependency manifest contains no exact versions, hashes, or lockfile references. Every installation resolves packages and transitive dependencies according to the package index state at installation time. This makes the build non-reproducible and allows future compromised, malicious, or incompatible releases to enter the runtime without any repository change. Several dependencies process network input, media, PDFs, model output, or execute complex agent workflows, increasing the consequences of an unsafe update. The audit did not establish that any currently named package is malicious. The confirmed weakness is the unrestricted resolution and installation process. ### Attack Path 1. A user runs the documented setup command. 2. `uv pip install` resolves the newest available versions allowed by the environment. 3. A dependency or transitive dependency has been compromised, maliciously transferred, or replaced by an unsafe release. 4. Installation executes package build hooks or installs hostile runtime code. 5. The package later runs inside TeamClaw and obtains the service account's access to secrets, user data, and network resources. ### Impact Assessment A dependency compromise can affect the complete TeamClaw process environment, inc ...[truncated 331 chars]
Remediation
<![CDATA[ ## Remediation Suggestions - Generate and commit a lockfile containing exact direct and transitive versions. - Require cryptographic hashes for every downloaded distribution. - Install with hash verification enabled. - Use a controlled package index or internal mirror. - Review package ownership, release history, build configuration, and transitive dependencies. - Separate development, optional bot, media, and production dependencies. - Run automated vulnerability and license scans against the locked dependency graph. - Update dependencies through reviewed pull requests rather than resolving mutable versions during deployment. - Build and test deployment artifacts once, then promote the same verified artifact between environments. ]]>
Vulnerability Patterns
  • Taint TrackingDirect Taint Flow, Variable-Mediated Taint Flow, Credential Exfiltration Chain
  • Prompt InjectionInstruction Override, Hidden Instructions, Exfiltration Commands
  • Data ExfiltrationExternal Transmission, Env Variable Harvesting, File System Enumeration
  • Privilege EscalationExcessive Permissions, Sudo/Root Execution, Credential Access
  • Supply ChainUnpinned Dependencies, External Script Fetching, Obfuscated Code
Findings (449)

Tainted flow: 'LOCAL_OPENAI_COMPLETIONS_URL' from os.getenv (line 28, credential/environment) → requests.post (network output)

Critical
Category
Data Flow
Content
# 直接透传请求体和 Authorization header 到后端
    auth_header = request.headers.get("Authorization", "")
    try:
        r = requests.post(
            LOCAL_OPENAI_COMPLETIONS_URL,
            json=request.get_json(silent=True),
            headers={
Confidence
90% confidence
Finding
Credentials or environment variables flow to a network sink. This is a high-confidence indicator of credential exfiltration.

Tainted flow: 'PORT_AGENT' from os.getenv (line 16, credential/environment) → requests.get (network output)

Critical
Category
Data Flow
Content
def proxy_openai_models():
    """透传 /v1/models"""
    try:
        r = requests.get(f"http://127.0.0.1:{PORT_AGENT}/v1/models", timeout=10)
        return Response(r.content, status=r.status_code, content_type=r.headers.get("content-type", "application/json"))
    except Exception as e:
        return jsonify({"error": str(e)}), 500
Confidence
90% confidence
Finding
Credentials or environment variables flow to a network sink. This is a high-confidence indicator of credential exfiltration.

Tainted flow: 'LOCAL_LOGIN_URL' from os.getenv (line 20, credential/environment) → requests.post (network output)

Critical
Category
Data Flow
Content
password = request.json.get("password", "")

    try:
        r = requests.post(LOCAL_LOGIN_URL, json={"user_id": user_id, "password": password}, timeout=10)
        if r.status_code == 200:
            # 登录成功,在 Flask session 中记录
            session["user_id"] = user_id
Confidence
90% confidence
Finding
Credentials or environment variables flow to a network sink. This is a high-confidence indicator of credential exfiltration.

Tainted flow: 'LOCAL_OPENAI_COMPLETIONS_URL' from os.getenv (line 28, credential/environment) → requests.post (network output)

Critical
Category
Data Flow
Content
}

    try:
        r = requests.post(
            LOCAL_OPENAI_COMPLETIONS_URL,
            json=openai_payload,
            headers={"Authorization": f"Bearer {user_id}:{password}"},
Confidence
90% confidence
Finding
Credentials or environment variables flow to a network sink. This is a high-confidence indicator of credential exfiltration.

Tainted flow: 'LOCAL_OPENAI_COMPLETIONS_URL' from os.getenv (line 28, credential/environment) → requests.post (network output)

Critical
Category
Data Flow
Content
}

    try:
        r = requests.post(
            LOCAL_OPENAI_COMPLETIONS_URL,
            json=openai_payload,
            headers={"Authorization": f"Bearer {user_id}:{password}"},
Confidence
90% confidence
Finding
Credentials or environment variables flow to a network sink. This is a high-confidence indicator of credential exfiltration.

Tainted flow: 'LOCAL_AGENT_CANCEL_URL' from os.getenv (line 19, credential/environment) → requests.post (network output)

Critical
Category
Data Flow
Content
return jsonify({"error": "未登录"}), 401
    session_id = request.json.get("session_id", "default") if request.is_json else "default"
    try:
        r = requests.post(LOCAL_AGENT_CANCEL_URL, json={"user_id": user_id, "password": password, "session_id": session_id}, timeout=5)
        return jsonify(r.json())
    except Exception as e:
        return jsonify({"error": str(e)}), 500
Confidence
90% confidence
Finding
Credentials or environment variables flow to a network sink. This is a high-confidence indicator of credential exfiltration.

Tainted flow: 'LOCAL_TTS_URL' from os.getenv (line 25, credential/environment) → requests.post (network output)

Critical
Category
Data Flow
Content
payload = {"user_id": user_id, "password": password, "text": text}
        if voice:
            payload["voice"] = voice
        r = requests.post(LOCAL_TTS_URL, json=payload, timeout=60)
        if r.status_code != 200:
            return jsonify({"error": f"TTS 服务错误: {r.status_code}"}), r.status_code
Confidence
90% confidence
Finding
Credentials or environment variables flow to a network sink. This is a high-confidence indicator of credential exfiltration.

Tainted flow: 'LOCAL_TOOLS_URL' from os.getenv (line 21, credential/environment) → requests.get (network output)

Critical
Category
Data Flow
Content
def proxy_tools():
    """代理获取工具列表请求到后端 Agent"""
    try:
        r = requests.get(LOCAL_TOOLS_URL, headers={"X-Internal-Token": INTERNAL_TOKEN}, timeout=10)
        return jsonify(r.json())
    except Exception as e:
        return jsonify({"error": str(e), "tools": []}), 500
Confidence
90% confidence
Finding
Credentials or environment variables flow to a network sink. This is a high-confidence indicator of credential exfiltration.

Tainted flow: 'LOCAL_SESSIONS_URL' from os.getenv (line 22, credential/environment) → requests.post (network output)

Critical
Category
Data Flow
Content
if not user_id or not password:
        return jsonify({"error": "未登录"}), 401
    try:
        r = requests.post(LOCAL_SESSIONS_URL, json={"user_id": user_id, "password": password}, timeout=15)
        return jsonify(r.json()), r.status_code
    except Exception as e:
        return jsonify({"error": str(e)}), 500
Confidence
90% confidence
Finding
Credentials or environment variables flow to a network sink. This is a high-confidence indicator of credential exfiltration.

Tainted flow: 'PORT_AGENT' from os.getenv (line 16, credential/environment) → requests.post (network output)

Critical
Category
Data Flow
Content
if not user_id or not password:
        return jsonify({"error": "未登录"}), 401
    try:
        r = requests.post(
            f"http://127.0.0.1:{PORT_AGENT}/sessions_status",
            json={"user_id": user_id, "password": password},
            timeout=5,
Confidence
90% confidence
Finding
Credentials or environment variables flow to a network sink. This is a high-confidence indicator of credential exfiltration.

Tainted flow: 'OASIS_BASE_URL' from os.getenv (line 33, credential/environment) → requests.get (network output)

Critical
Category
Data Flow
Content
"""Proxy to fetch OpenClaw session list from OASIS server."""
    filter_kw = request.args.get("filter", "")
    try:
        r = requests.get(
            f"{OASIS_BASE_URL}/sessions/openclaw",
            params={"filter": filter_kw},
            timeout=10,
Confidence
90% confidence
Finding
Credentials or environment variables flow to a network sink. This is a high-confidence indicator of credential exfiltration.

Tainted flow: 'LOCAL_SESSION_HISTORY_URL' from os.getenv (line 23, credential/environment) → requests.post (network output)

Critical
Category
Data Flow
Content
return jsonify({"error": "未登录"}), 401
    sid = request.json.get("session_id", "")
    try:
        r = requests.post(LOCAL_SESSION_HISTORY_URL, json={
            "user_id": user_id, "password": password, "session_id": sid
        }, timeout=15)
        return jsonify(r.json()), r.status_code
Confidence
90% confidence
Finding
Credentials or environment variables flow to a network sink. This is a high-confidence indicator of credential exfiltration.

Tainted flow: 'LOCAL_SESSION_STATUS_URL' from os.getenv (line 26, credential/environment) → requests.post (network output)

Critical
Category
Data Flow
Content
return jsonify({"has_new_messages": False}), 200
    sid = request.json.get("session_id", "") if request.is_json else ""
    try:
        r = requests.post(LOCAL_SESSION_STATUS_URL, json={
            "user_id": user_id, "password": password, "session_id": sid
        }, timeout=5)
        return jsonify(r.json()), r.status_code
Confidence
90% confidence
Finding
Credentials or environment variables flow to a network sink. This is a high-confidence indicator of credential exfiltration.

Tainted flow: 'LOCAL_DELETE_SESSION_URL' from os.getenv (line 24, credential/environment) → requests.post (network output)

Critical
Category
Data Flow
Content
return jsonify({"error": "未登录"}), 401
    sid = request.json.get("session_id", "") if request.is_json else ""
    try:
        r = requests.post(LOCAL_DELETE_SESSION_URL, json={
            "user_id": user_id, "password": password, "session_id": sid
        }, timeout=15)
        return jsonify(r.json()), r.status_code
Confidence
90% confidence
Finding
Credentials or environment variables flow to a network sink. This is a high-confidence indicator of credential exfiltration.

Tainted flow: 'PORT_AGENT' from os.getenv (line 16, credential/environment) → requests.get (network output)

Critical
Category
Data Flow
Content
if not uid:
        return jsonify([]), 200
    try:
        r = requests.get(f"http://127.0.0.1:{PORT_AGENT}/groups", headers=headers, timeout=10)
        return jsonify(r.json()), r.status_code
    except Exception as e:
        return jsonify({"error": str(e)}), 500
Confidence
90% confidence
Finding
Credentials or environment variables flow to a network sink. This is a high-confidence indicator of credential exfiltration.

Tainted flow: 'PORT_AGENT' from os.getenv (line 16, credential/environment) → requests.post (network output)

Critical
Category
Data Flow
Content
return jsonify({"error": "未登录"}), 401
    try:
        headers["Content-Type"] = "application/json"
        r = requests.post(f"http://127.0.0.1:{PORT_AGENT}/groups", json=request.get_json(silent=True), headers=headers, timeout=10)
        return jsonify(r.json()), r.status_code
    except Exception as e:
        return jsonify({"error": str(e)}), 500
Confidence
90% confidence
Finding
Credentials or environment variables flow to a network sink. This is a high-confidence indicator of credential exfiltration.

Tainted flow: 'PORT_AGENT' from os.getenv (line 16, credential/environment) → requests.get (network output)

Critical
Category
Data Flow
Content
if not uid:
        return jsonify({"error": "未登录"}), 401
    try:
        r = requests.get(f"http://127.0.0.1:{PORT_AGENT}/groups/{group_id}", headers=headers, timeout=10)
        return jsonify(r.json()), r.status_code
    except Exception as e:
        return jsonify({"error": str(e)}), 500
Confidence
90% confidence
Finding
Credentials or environment variables flow to a network sink. This is a high-confidence indicator of credential exfiltration.

Tainted flow: 'PORT_AGENT' from os.getenv (line 16, credential/environment) → requests.put (network output)

Critical
Category
Data Flow
Content
return jsonify({"error": "未登录"}), 401
    try:
        headers["Content-Type"] = "application/json"
        r = requests.put(f"http://127.0.0.1:{PORT_AGENT}/groups/{group_id}", json=request.get_json(silent=True), headers=headers, timeout=10)
        return jsonify(r.json()), r.status_code
    except Exception as e:
        return jsonify({"error": str(e)}), 500
Confidence
90% confidence
Finding
Credentials or environment variables flow to a network sink. This is a high-confidence indicator of credential exfiltration.

Tainted flow: 'PORT_AGENT' from os.getenv (line 16, credential/environment) → requests.get (network output)

Critical
Category
Data Flow
Content
return jsonify({"messages": []}), 200
    try:
        after_id = request.args.get("after_id", "0")
        r = requests.get(
            f"http://127.0.0.1:{PORT_AGENT}/groups/{group_id}/messages",
            params={"after_id": after_id},
            headers=headers, timeout=10,
Confidence
90% confidence
Finding
Credentials or environment variables flow to a network sink. This is a high-confidence indicator of credential exfiltration.

Tainted flow: 'PORT_AGENT' from os.getenv (line 16, credential/environment) → requests.post (network output)

Critical
Category
Data Flow
Content
return jsonify({"error": "未登录"}), 401
    try:
        headers["Content-Type"] = "application/json"
        r = requests.post(
            f"http://127.0.0.1:{PORT_AGENT}/groups/{group_id}/messages",
            json=request.get_json(silent=True),
            headers=headers, timeout=10,
Confidence
90% confidence
Finding
Credentials or environment variables flow to a network sink. This is a high-confidence indicator of credential exfiltration.

Tainted flow: 'PORT_AGENT' from os.getenv (line 16, credential/environment) → requests.post (network output)

Critical
Category
Data Flow
Content
if not uid:
        return jsonify({"error": "未登录"}), 401
    try:
        r = requests.post(
            f"http://127.0.0.1:{PORT_AGENT}/groups/{group_id}/mute",
            headers=headers, timeout=10,
        )
Confidence
90% confidence
Finding
Credentials or environment variables flow to a network sink. This is a high-confidence indicator of credential exfiltration.

Tainted flow: 'PORT_AGENT' from os.getenv (line 16, credential/environment) → requests.post (network output)

Critical
Category
Data Flow
Content
if not uid:
        return jsonify({"error": "未登录"}), 401
    try:
        r = requests.post(
            f"http://127.0.0.1:{PORT_AGENT}/groups/{group_id}/unmute",
            headers=headers, timeout=10,
        )
Confidence
90% confidence
Finding
Credentials or environment variables flow to a network sink. This is a high-confidence indicator of credential exfiltration.

Tainted flow: 'PORT_AGENT' from os.getenv (line 16, credential/environment) → requests.get (network output)

Critical
Category
Data Flow
Content
if not uid:
        return jsonify({"muted": False}), 200
    try:
        r = requests.get(
            f"http://127.0.0.1:{PORT_AGENT}/groups/{group_id}/mute_status",
            headers=headers, timeout=10,
        )
Confidence
90% confidence
Finding
Credentials or environment variables flow to a network sink. This is a high-confidence indicator of credential exfiltration.

Tainted flow: 'PORT_AGENT' from os.getenv (line 16, credential/environment) → requests.get (network output)

Critical
Category
Data Flow
Content
if not uid:
        return jsonify({"sessions": []}), 200
    try:
        r = requests.get(
            f"http://127.0.0.1:{PORT_AGENT}/groups/{group_id}/sessions",
            headers=headers, timeout=15,
        )
Confidence
90% confidence
Finding
Credentials or environment variables flow to a network sink. This is a high-confidence indicator of credential exfiltration.

Tainted flow: 'OASIS_BASE_URL' from os.getenv (line 33, credential/environment) → requests.get (network output)

Critical
Category
Data Flow
Content
return jsonify([]), 200
    try:
        print(f"[OASIS Proxy] Fetching topics from {OASIS_BASE_URL}/topics for user={user_id}")
        r = requests.get(f"{OASIS_BASE_URL}/topics", params={"user_id": user_id}, timeout=10)
        print(f"[OASIS Proxy] Response status: {r.status_code}, count: {len(r.json()) if r.text else 0}")
        return jsonify(r.json()), r.status_code
    except Exception as e:
Confidence
90% confidence
Finding
Credentials or environment variables flow to a network sink. This is a high-confidence indicator of credential exfiltration.