diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index fc3a43e..4ffaa1e 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -69,7 +69,9 @@ jobs: - name: Install test deps run: | python -m pip install --upgrade pip - python -m pip install numpy pillow + # Keep aligned with local pre-push/full-test scripts. + # aiohttp is required by multiple unit-test import paths. + python -m pip install numpy pillow aiohttp - name: Run unit tests env: diff --git a/api/routes.py b/api/routes.py index f205cef..49097c3 100644 --- a/api/routes.py +++ b/api/routes.py @@ -67,7 +67,10 @@ if web is not None: # CRITICAL: These imports MUST remain present. # If edited out, module-level placeholders stay as None and handlers raise at runtime # (e.g., TypeError: 'NoneType' object is not callable), producing noisy aiohttp tracebacks. - from ..services.access_control import require_observability_access + from ..services.access_control import ( + require_admin_token, + require_observability_access, + ) from ..services.log_tail import tail_log from ..services.metrics import metrics from ..services.rate_limit import check_rate_limit @@ -109,6 +112,7 @@ if web is not None: # IMPORTANT: keep PACK_* imports aligned with config.py (VERSION/config_path do not exist). from config import LOG_FILE, PACK_NAME, PACK_START_TIME, PACK_VERSION + from services.access_control import require_admin_token # type: ignore from services.access_control import require_observability_access # type: ignore from services.log_tail import tail_log # type: ignore from services.metrics import metrics # type: ignore @@ -261,8 +265,8 @@ async def logs_tail_handler(request: web.Request) -> web.Response: ok, init_error = _ensure_observability_deps_ready() if not ok: return web.json_response({"ok": False, "error": init_error}, status=500) - # S14: Access Control - allowed, error = require_observability_access(request) + # S34: Trace/Log data is high sensitivity -> Require Admin Token + allowed, error = require_admin_token(request) if not allowed: return web.json_response({"ok": False, "error": error}, status=403) @@ -364,7 +368,8 @@ async def trace_handler(request: web.Request) -> web.Response: ok, init_error = _ensure_observability_deps_ready() if not ok: return web.json_response({"ok": False, "error": init_error}, status=500) - allowed, error = require_observability_access(request) + # S34: Trace/Log data is high sensitivity -> Require Admin Token + allowed, error = require_admin_token(request) if not allowed: return web.json_response({"ok": False, "error": error}, status=403) diff --git a/connector/platforms/kakao_webhook.py b/connector/platforms/kakao_webhook.py index fbf9c98..6be2ad7 100644 --- a/connector/platforms/kakao_webhook.py +++ b/connector/platforms/kakao_webhook.py @@ -39,6 +39,41 @@ def _import_aiohttp_web(): return aiohttp, web +class _CompatResponse: + """Minimal response shim for unit tests when aiohttp is unavailable.""" + + def __init__( + self, + *, + status: int = 200, + text: str = "", + content_type: str = "text/plain", + body: Optional[bytes] = None, + ): + self.status = status + self.text = text + self.content_type = content_type + self.body = body if body is not None else text.encode("utf-8") + + +def _make_response(web, *, status: int = 200, text: str = "OK"): + if web is not None: + return web.Response(status=status, text=text) + return _CompatResponse(status=status, text=text) + + +def _make_json_response(web, data: dict, *, status: int = 200): + body = json.dumps(data, ensure_ascii=False).encode("utf-8") + if web is not None: + return web.json_response(data, status=status) + return _CompatResponse( + status=status, + text=body.decode("utf-8"), + content_type="application/json", + body=body, + ) + + class KakaoWebhookServer: """ KakaoTalk payload adapter. @@ -112,14 +147,15 @@ class KakaoWebhookServer: async def handle_webhook(self, request): """POST handler for Kakao Skill payloads.""" _, web = _import_aiohttp_web() - if web is None: - raise RuntimeError("aiohttp not available") + # IMPORTANT: + # CI unit tests call this handler directly without aiohttp installed. + # Keep this path runnable; do not replace with a hard RuntimeError. try: body_bytes = await request.read() payload = json.loads(body_bytes) except json.JSONDecodeError: - return web.Response(status=400, text="Bad JSON") + return _make_response(web, status=400, text="Bad JSON") # S32: Replay Guard (Content Hash Dedup) # We use a hash of the body bytes as the "nonce" for deduplication. @@ -127,9 +163,8 @@ class KakaoWebhookServer: content_hash = hashlib.sha256(body_bytes).hexdigest() if not self._replay_guard.check_and_record(content_hash): logger.warning(f"Replay rejected for Kakao hash: {content_hash}") - return web.Response( - status=200, text="OK" - ) # Return 200 to stop Kakao retries + # Return 200 to stop Kakao retries + return _make_response(web, status=200, text="OK") # Normalization # userRequest.user.id is the opaque user ID (botUserKey) @@ -169,8 +204,15 @@ class KakaoWebhookServer: try: resp = await self.router.handle(req) - if resp.text: - return self._build_text_response(resp.text) + # IMPORTANT: + # Router mocks in unit tests may return non-string `.text` values. + # Normalize defensively to avoid turning a valid routing flow into + # a JSON serialization error path. + resp_text = getattr(resp, "text", "") + if not isinstance(resp_text, str): + resp_text = str(resp_text) if resp_text is not None else "" + if resp_text: + return self._build_text_response(resp_text) else: # No response content (e.g. valid command but no output intended?) # Kakao requires *some* response payload or it treats as error. @@ -192,7 +234,7 @@ class KakaoWebhookServer: "version": "2.0", "template": {"outputs": [{"simpleText": {"text": text}}]}, } - return web.json_response(resp_data) + return _make_json_response(web, resp_data) def _build_error_response(self, error_msg: str): """Build simple error text response.""" diff --git a/connector/platforms/wechat_webhook.py b/connector/platforms/wechat_webhook.py index 13f32a0..6129523 100644 --- a/connector/platforms/wechat_webhook.py +++ b/connector/platforms/wechat_webhook.py @@ -47,6 +47,35 @@ def _import_aiohttp_web(): return aiohttp, web +class _CompatResponse: + """Minimal response shim for unit tests when aiohttp is unavailable.""" + + def __init__( + self, + *, + status: int = 200, + text: str = "", + content_type: str = "text/plain", + body: Optional[bytes] = None, + ): + self.status = status + self.text = text + self.content_type = content_type + self.body = body if body is not None else text.encode("utf-8") + + +def _make_response( + web, + *, + status: int = 200, + text: str = "", + content_type: str = "text/plain", +): + if web is not None: + return web.Response(status=status, text=text, content_type=content_type) + return _CompatResponse(status=status, text=text, content_type=content_type) + + # --------------------------------------------------------------------------- # R74 — Protocol constants # --------------------------------------------------------------------------- @@ -343,8 +372,9 @@ class WeChatWebhookServer: Must return echostr as plain text if signature is valid. """ _, web = _import_aiohttp_web() - if web is None: - raise RuntimeError("aiohttp not available") + # IMPORTANT: + # CI unit tests invoke handler logic directly without aiohttp installed. + # Do not hard-raise here; return compat responses so security logic remains testable. signature = request.query.get("signature", "") timestamp = request.query.get("timestamp", "") @@ -355,10 +385,10 @@ class WeChatWebhookServer: if verify_wechat_signature(token, timestamp, nonce, signature): logger.info("WeChat webhook verification succeeded") - return web.Response(text=echostr, content_type="text/plain") + return _make_response(web, text=echostr, content_type="text/plain") logger.warning("WeChat webhook verification failed") - return web.Response(status=403, text="Verification failed") + return _make_response(web, status=403, text="Verification failed") # ------------------------------------------------------------------ # POST — Inbound Messages (R74 + S31) @@ -367,8 +397,9 @@ class WeChatWebhookServer: async def handle_webhook(self, request): """POST handler for WeChat XML messages/events.""" _, web = _import_aiohttp_web() - if web is None: - raise RuntimeError("aiohttp not available") + # IMPORTANT: + # Keep handler behavior testable in environments without aiohttp. + # Server startup still requires aiohttp, but direct handler unit tests should not crash. # S31: Signature verification signature = request.query.get("signature", "") @@ -378,12 +409,12 @@ class WeChatWebhookServer: if not verify_wechat_signature(token, timestamp, nonce, signature): logger.warning("Invalid WeChat POST signature") - return web.Response(status=401, text="Invalid Signature") + return _make_response(web, status=401, text="Invalid Signature") # S31: Replay protection — nonce dedup if nonce and not self._replay_guard.check_and_record(nonce): logger.warning(f"Replay rejected for WeChat nonce: {nonce}") - return web.Response(status=403, text="Replay Rejected") + return _make_response(web, status=403, text="Replay Rejected") # S31: Timestamp freshness try: @@ -394,7 +425,7 @@ class WeChatWebhookServer: age_sec = now - ts_val if age_sec > self.REPLAY_WINDOW_SEC or age_sec < -60: logger.warning(f"Stale WeChat request: age={age_sec}s") - return web.Response(status=403, text="Stale Request") + return _make_response(web, status=403, text="Stale Request") # Read and parse XML with S31 budgets body_bytes = await request.read() @@ -403,13 +434,13 @@ class WeChatWebhookServer: fields = parse_wechat_xml(body_bytes) except XMLBudgetExceeded as e: logger.warning(f"WeChat XML budget exceeded: {e}") - return web.Response(status=400, text="Bad Request") + return _make_response(web, status=400, text="Bad Request") # R74: Normalize to canonical event event = normalize_wechat_event(fields) if event is None: # Unsupported message type — return empty success to WeChat - return web.Response(text="success", content_type="text/plain") + return _make_response(web, text="success", content_type="text/plain") # S31: Allowlist check (soft-deny) sender_id = event["sender_id"] @@ -429,7 +460,7 @@ class WeChatWebhookServer: # Per-message dedup (MsgId-based, distinct from nonce-based replay) if message_id and not self._replay_guard.check_and_record(f"msg:{message_id}"): logger.debug(f"Duplicate WeChat MsgId: {message_id}") - return web.Response(text="success", content_type="text/plain") + return _make_response(web, text="success", content_type="text/plain") req = CommandRequest( platform="wechat", @@ -448,14 +479,15 @@ class WeChatWebhookServer: to_user = event["sender_id"] from_user = event["to_user"] reply_xml = build_text_reply_xml(to_user, from_user, resp.text) - return web.Response( + return _make_response( + web, text=reply_xml, content_type="application/xml", ) except Exception as e: logger.exception(f"Error handling WeChat command: {e}") - return web.Response(text="success", content_type="text/plain") + return _make_response(web, text="success", content_type="text/plain") # ------------------------------------------------------------------ # Outbound: Text (Customer Service Message API) diff --git a/debug_s35_import.py b/debug_s35_import.py new file mode 100644 index 0000000..09c5304 --- /dev/null +++ b/debug_s35_import.py @@ -0,0 +1,32 @@ +""" +Debug script for S35 Transform Isolation. +Verifies that the correct executor (TransformProcessRunner) is allowed/loaded. +""" + +import os +import sys + +# Ensure project root is in path +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) + +from services.constrained_transforms import get_transform_executor +from services.transform_runner import TransformProcessRunner + + +def main(): + print("Checking S35 Transform Executor...") + executor = get_transform_executor() + print(f"Executor Type: {type(executor)}") + + if isinstance(executor, TransformProcessRunner): + print("SUCCESS: TransformProcessRunner is active (Isolation Enabled).") + else: + print("WARNING: TransformProcessRunner is NOT active.") + # Check if feature flag is enabled + from services.transform_common import is_transforms_enabled + + print(f"Feature Flag Enabled: {is_transforms_enabled()}") + + +if __name__ == "__main__": + main() diff --git a/scripts/pre_push_checks.sh b/scripts/pre_push_checks.sh index 0d719b1..9389224 100644 --- a/scripts/pre_push_checks.sh +++ b/scripts/pre_push_checks.sh @@ -40,13 +40,50 @@ require_cmd() { fi } +pip_install_or_fail() { + local why="$1" + shift + if "$VENV_PY" -m pip install "$@"; then + return 0 + fi + echo "[pre-push] ERROR: failed to install dependency ($why): $*" >&2 + exit 1 +} + +is_wsl() { + grep -qiE "(microsoft|wsl)" /proc/version 2>/dev/null +} + +select_venv_dir() { + # Explicit override for advanced/local setups. + if [ -n "${OPENCLAW_TEST_VENV:-}" ]; then + echo "$OPENCLAW_TEST_VENV" + return 0 + fi + + case "$UNAME_S" in + MINGW*|MSYS*|CYGWIN*) + echo "$ROOT_DIR/.venv" + ;; + *) + # IMPORTANT: + # In WSL, prefer dedicated Linux venv to avoid mixing with Windows .venv. + if is_wsl; then + echo "$ROOT_DIR/.venv-wsl" + else + echo "$ROOT_DIR/.venv" + fi + ;; + esac +} + resolve_venv_python() { case "$UNAME_S" in MINGW*|MSYS*|CYGWIN*) - echo "$ROOT_DIR/.venv/Scripts/python.exe" + echo "$VENV_DIR/Scripts/python.exe" ;; *) - echo "$ROOT_DIR/.venv/bin/python" + echo "$VENV_DIR/bin/python" ;; esac } @@ -70,24 +107,24 @@ bootstrap_venv() { fi if [ -e "$venv_py" ]; then - echo "[pre-push] WARN: existing .venv is invalid; recreating with a Windows-native Python." >&2 - rm -rf "$ROOT_DIR/.venv" + echo "[pre-push] WARN: existing venv is invalid; recreating: $VENV_DIR" >&2 + rm -rf "$VENV_DIR" fi - echo "[pre-push] INFO: creating project .venv ..." >&2 + echo "[pre-push] INFO: creating project venv at $VENV_DIR ..." >&2 case "$UNAME_S" in MINGW*|MSYS*|CYGWIN*) # CRITICAL: on Git Bash, `python3` may resolve to MSYS `/usr/bin/python`, # which creates a broken Windows venv (`No Python at "/usr/bin\python.exe"`). # Always prefer Windows-native launchers/interpreters. if command -v py.exe >/dev/null 2>&1; then - py.exe -3 -m venv "$ROOT_DIR/.venv" + py.exe -3 -m venv "$VENV_DIR" elif [ -x "/c/Windows/py.exe" ]; then - /c/Windows/py.exe -3 -m venv "$ROOT_DIR/.venv" + /c/Windows/py.exe -3 -m venv "$VENV_DIR" elif command -v python.exe >/dev/null 2>&1; then - python.exe -m venv "$ROOT_DIR/.venv" + python.exe -m venv "$VENV_DIR" elif command -v py >/dev/null 2>&1; then - py -3 -m venv "$ROOT_DIR/.venv" + py -3 -m venv "$VENV_DIR" else echo "[pre-push] ERROR: no Windows Python launcher found (py.exe/python.exe)." >&2 exit 1 @@ -95,9 +132,9 @@ bootstrap_venv() { ;; *) if command -v python3 >/dev/null 2>&1; then - python3 -m venv "$ROOT_DIR/.venv" + python3 -m venv "$VENV_DIR" elif command -v python >/dev/null 2>&1; then - python -m venv "$ROOT_DIR/.venv" + python -m venv "$VENV_DIR" else echo "[pre-push] ERROR: no bootstrap Python found (python3/python)." >&2 exit 1 @@ -106,7 +143,7 @@ bootstrap_venv() { esac if ! is_venv_python_healthy "$venv_py"; then - echo "[pre-push] ERROR: failed to initialize project .venv." >&2 + echo "[pre-push] ERROR: failed to initialize project venv: $VENV_DIR" >&2 exit 1 fi echo "$venv_py" @@ -116,19 +153,33 @@ pre_commit_cmd() { "$VENV_PY" -m pre_commit "$@" } -# CRITICAL: pre-push must always run pre-commit from project .venv. +# CRITICAL: pre-push must always run pre-commit from project venv. # Do not switch this back to global `pre-commit` command lookup. # This prevents mixed global/user installs from hijacking hook execution. +VENV_DIR="$(select_venv_dir)" VENV_PY="$(bootstrap_venv)" if ! "$VENV_PY" -m pre_commit --version >/dev/null 2>&1; then - echo "[pre-push] INFO: installing pre-commit into project .venv ..." >&2 - "$VENV_PY" -m pip install -U pip pre-commit + echo "[pre-push] INFO: installing pre-commit into project venv ($VENV_DIR) ..." >&2 + pip_install_or_fail "required for pre-commit hooks" -U pip pre-commit fi if ! "$VENV_PY" -c "import black" >/dev/null 2>&1; then # Keep black in the same interpreter used by local black-single hook. - echo "[pre-push] INFO: installing black into project .venv ..." >&2 - "$VENV_PY" -m pip install black==24.1.1 + echo "[pre-push] INFO: installing black into project venv ($VENV_DIR) ..." >&2 + pip_install_or_fail "required by black-single hook" black==24.1.1 fi + +# IMPORTANT: +# Pre-push now runs backend unit tests. Keep minimal runtime deps aligned with +# CI unit-test job to avoid "passes locally, fails on GitHub" drift. +if ! "$VENV_PY" -c "import numpy, PIL" >/dev/null 2>&1; then + echo "[pre-push] INFO: installing numpy/pillow into project venv ($VENV_DIR) ..." >&2 + pip_install_or_fail "required by unit tests" numpy pillow +fi +if ! "$VENV_PY" -c "import aiohttp" >/dev/null 2>&1; then + echo "[pre-push] INFO: installing aiohttp into project venv ($VENV_DIR) ..." >&2 + pip_install_or_fail "required by unit tests/import paths" aiohttp +fi + require_cmd npm run_pre_commit_safe() { @@ -225,13 +276,17 @@ if [ "$NODE_MAJOR" -lt 18 ]; then fi echo "[pre-push] Node version: $(node -v)" -echo "[pre-push] 1/3 detect-secrets" +echo "[pre-push] 1/4 detect-secrets" run_pre_commit_safe run detect-secrets --all-files -echo "[pre-push] 2/3 pre-commit all hooks" +echo "[pre-push] 2/4 pre-commit all hooks" run_pre_commit_safe run --all-files -echo "[pre-push] 3/3 npm test (Playwright)" +echo "[pre-push] 3/4 backend unit tests" +MOLTBOT_STATE_DIR="$ROOT_DIR/moltbot_state/_pre_push_unit" \ + "$VENV_PY" scripts/run_unittests.py --start-dir tests --pattern "test_*.py" + +echo "[pre-push] 4/4 npm test (Playwright)" npm test echo "[pre-push] PASS" diff --git a/scripts/precommit_black_single.sh b/scripts/precommit_black_single.sh index 6362988..9da5017 100644 --- a/scripts/precommit_black_single.sh +++ b/scripts/precommit_black_single.sh @@ -6,30 +6,64 @@ ROOT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)" export BLACK_CACHE_DIR="${BLACK_CACHE_DIR:-$ROOT_DIR/.tmp/black-cache}" mkdir -p "$BLACK_CACHE_DIR" -# CRITICAL: Always prefer project-local .venv interpreter for Black. +# CRITICAL: Always prefer project-local venv interpreter for Black. # Without this, Windows can accidentally pick global Python (e.g. C:\Program Files\Python312) # where `black` is not installed, causing flaky pre-commit failures. # DO NOT change this back to "python/python3 from PATH first" unless you also # guarantee black is installed in every global interpreter used by contributors. + +is_wsl() { + grep -qiE "(microsoft|wsl)" /proc/version 2>/dev/null +} + can_use_python() { local candidate="$1" [ -f "$candidate" ] || return 1 "$candidate" -c "import sys; print(sys.executable)" >/dev/null 2>&1 } -has_project_venv=false -[ -d "$ROOT_DIR/.venv" ] && has_project_venv=true +select_venv_dirs() { + local dirs=() + if [ -n "${OPENCLAW_TEST_VENV:-}" ]; then + dirs+=("$OPENCLAW_TEST_VENV") + fi + if is_wsl; then + dirs+=("$ROOT_DIR/.venv-wsl") + fi + dirs+=("$ROOT_DIR/.venv") + printf '%s\n' "${dirs[@]}" +} -if can_use_python "$ROOT_DIR/.venv/Scripts/python.exe"; then - PY_CMD="$ROOT_DIR/.venv/Scripts/python.exe" -elif can_use_python "$ROOT_DIR/.venv/bin/python"; then - PY_CMD="$ROOT_DIR/.venv/bin/python" -elif [ "$has_project_venv" = true ]; then - # CRITICAL: if .venv exists but is broken, fail fast instead of silently - # falling back to a random global Python (which reintroduces flakiness). - # DO NOT relax this guard; it is intentional to prevent silent environment drift. - echo "ERROR: project .venv exists but Python is unusable. Recreate .venv and retry." >&2 - exit 1 +check_venv_dir() { + local dir="$1" + if can_use_python "$dir/Scripts/python.exe"; then + PY_CMD="$dir/Scripts/python.exe" + return 0 + fi + if can_use_python "$dir/bin/python"; then + PY_CMD="$dir/bin/python" + return 0 + fi + return 1 +} + +while IFS= read -r vdir; do + [ -n "$vdir" ] || continue + if [ -d "$vdir" ]; then + if check_venv_dir "$vdir"; then + break + fi + # IMPORTANT: + # If a preferred project venv exists but is unusable, fail fast instead of + # silently falling back to random global Python (prevents environment drift). + echo "ERROR: project venv exists but Python is unusable: $vdir" >&2 + echo "Recreate this venv and retry." >&2 + exit 1 + fi +done < <(select_venv_dirs) + +if [ -n "${PY_CMD:-}" ]; then + : elif command -v python >/dev/null 2>&1; then # Fallback chain only for environments that intentionally do not use .venv. PY_CMD="$(command -v python)" diff --git a/scripts/run_full_tests_linux.sh b/scripts/run_full_tests_linux.sh index 2160ea7..3418a17 100644 --- a/scripts/run_full_tests_linux.sh +++ b/scripts/run_full_tests_linux.sh @@ -22,6 +22,25 @@ require_cmd() { fi } +is_wsl() { + grep -qiE "(microsoft|wsl)" /proc/version 2>/dev/null +} + +select_venv_dir() { + # Explicit override for advanced/local setups. + if [ -n "${OPENCLAW_TEST_VENV:-}" ]; then + echo "$OPENCLAW_TEST_VENV" + return 0 + fi + # IMPORTANT: + # In WSL, prefer dedicated Linux venv to avoid clashing with Windows .venv. + if is_wsl; then + echo "$ROOT_DIR/.venv-wsl" + else + echo "$ROOT_DIR/.venv" + fi +} + pip_install_or_fail() { local why="$1" shift @@ -30,7 +49,7 @@ pip_install_or_fail() { fi echo "[tests] ERROR: failed to install dependency ($why): $*" >&2 echo "[tests] HINT: check internet/proxy, then retry the script." >&2 - echo "[tests] HINT: if offline, pre-install into .venv manually: $VENV_PY -m pip install $*" >&2 + echo "[tests] HINT: if offline, pre-install into venv manually: $VENV_PY -m pip install $*" >&2 exit 1 } @@ -38,13 +57,14 @@ require_cmd node require_cmd npm # Always use project-local venv to avoid global interpreter / tool drift. -VENV_PY="$ROOT_DIR/.venv/bin/python" +VENV_DIR="$(select_venv_dir)" +VENV_PY="$VENV_DIR/bin/python" if [ ! -x "$VENV_PY" ]; then - echo "[tests] Creating project venv at $ROOT_DIR/.venv ..." + echo "[tests] Creating project venv at $VENV_DIR ..." if command -v python3 >/dev/null 2>&1; then - python3 -m venv "$ROOT_DIR/.venv" + python3 -m venv "$VENV_DIR" elif command -v python >/dev/null 2>&1; then - python -m venv "$ROOT_DIR/.venv" + python -m venv "$VENV_DIR" else echo "[tests] ERROR: no bootstrap Python found (need python3 or python)" >&2 exit 1 @@ -52,12 +72,12 @@ if [ ! -x "$VENV_PY" ]; then fi if ! "$VENV_PY" -m pre_commit --version >/dev/null 2>&1; then - echo "[tests] Installing pre-commit into project venv ..." + echo "[tests] Installing pre-commit into project venv ($VENV_DIR) ..." pip_install_or_fail "required for detect-secrets and hook validation" -U pip pre-commit fi if ! "$VENV_PY" -c "import aiohttp" >/dev/null 2>&1; then - echo "[tests] Installing aiohttp into project venv ..." + echo "[tests] Installing aiohttp into project venv ($VENV_DIR) ..." pip_install_or_fail "required by import paths used in unit tests" aiohttp fi diff --git a/services/approvals/storage.py b/services/approvals/storage.py index 6662c63..a001112 100644 --- a/services/approvals/storage.py +++ b/services/approvals/storage.py @@ -57,16 +57,19 @@ def _atomic_write(path: str, data: Dict) -> None: raise +from ..integrity import IntegrityError, load_verified, save_verified + + def load_approvals() -> Dict[str, ApprovalRequest]: - """Load approvals from disk.""" + """Load approvals from disk with integrity check.""" path = _get_approvals_path() if not os.path.exists(path): return {} try: - with open(path, "r", encoding="utf-8") as f: - data = json.load(f) + # R77: Load verification + data = load_verified(path, expected_version=1, migrate=True) result = {} if isinstance(data, dict) and "approvals" in data: @@ -79,13 +82,20 @@ def load_approvals() -> Dict[str, ApprovalRequest]: logger.info(f"Loaded {len(result)} approval records") return result - except (json.JSONDecodeError, OSError) as e: + except IntegrityError as e: + # R77: Fail-closed logic with escalation + logger.critical( + f"R77: Integrity violation detected in approvals file {path}: {e}" + ) + # Return empty (deny all pending approvals) which is safe fail-state + return {} + except Exception as e: logger.error(f"Failed to load approvals: {e}") return {} def save_approvals(approvals: Dict[str, ApprovalRequest]) -> bool: - """Save approvals to disk.""" + """Save approvals to disk with integrity envelope.""" path = _get_approvals_path() try: @@ -94,7 +104,8 @@ def save_approvals(approvals: Dict[str, ApprovalRequest]) -> bool: "saved_at": datetime.now(timezone.utc).isoformat(), "approvals": [a.to_dict() for a in approvals.values()], } - _atomic_write(path, data) + # R77: Atomic verified save + save_verified(path, data, version=1) return True except Exception as e: logger.error(f"Failed to save approvals: {e}") diff --git a/services/checkpoints.py b/services/checkpoints.py index 4f84f4f..1da1ecd 100644 --- a/services/checkpoints.py +++ b/services/checkpoints.py @@ -18,6 +18,8 @@ except ImportError: # Fallback for tests or decoupled run DATA_DIR = os.path.join(os.path.dirname(os.path.dirname(__file__)), "data") +from .integrity import IntegrityError, load_verified, save_verified + logger = logging.getLogger("ComfyUI-OpenClaw.services.checkpoints") CHECKPOINTS_DIR = os.path.join(DATA_DIR, "checkpoints") @@ -45,9 +47,14 @@ def list_checkpoints() -> List[Dict[str, Any]]: for entry in it: if entry.name.endswith(".meta.json") and entry.is_file(): try: - with open(entry.path, "r", encoding="utf-8") as f: - meta = json.load(f) - checkpoints.append(meta) + + meta = load_verified(entry.path, migrate=True) + checkpoints.append(meta) + except IntegrityError as e: + logger.critical( + f"R77: Integrity violation in checkpoint {entry.name}: {e}" + ) + # Skip this file (fail-closed for this item) except Exception: logger.warning(f"Failed to read checkpoint meta: {entry.name}") except OSError: @@ -66,12 +73,14 @@ def get_checkpoint(checkpoint_id: str) -> Optional[Dict[str, Any]]: return None try: - with open(meta_path, "r", encoding="utf-8") as f: - meta = json.load(f) - with open(payload_path, "r", encoding="utf-8") as f: - workflow = json.load(f) + + meta = load_verified(meta_path, migrate=True) + workflow = load_verified(payload_path, migrate=True) return {"id": checkpoint_id, "meta": meta, "workflow": workflow} + except IntegrityError as e: + logger.critical(f"R77: Integrity violation in checkpoint {checkpoint_id}: {e}") + return None except Exception as e: logger.error(f"Error reading checkpoint {checkpoint_id}: {e}") return None @@ -148,8 +157,8 @@ def create_checkpoint( meta_path, payload_path = _get_paths(cid) try: - _atomic_write(meta_path, json.dumps(meta, indent=2)) - _atomic_write(payload_path, workflow_json) + save_verified(meta_path, meta) + save_verified(payload_path, workflow) except Exception as e: # Cleanup on fail (although atomic write minimizes this risk for individual files) if os.path.exists(meta_path): diff --git a/services/constrained_transforms.py b/services/constrained_transforms.py index a5aa06e..a846451 100644 --- a/services/constrained_transforms.py +++ b/services/constrained_transforms.py @@ -16,311 +16,55 @@ Default posture: DISABLED. Requires OPENCLAW_ENABLE_TRANSFORMS=1. from __future__ import annotations -import hashlib import importlib import importlib.util import json import logging import os -import signal -import sys import threading import time -from dataclasses import asdict, dataclass, field -from enum import Enum -from pathlib import Path -from typing import Any, Callable, Dict, List, Optional, Set +from typing import Any, Dict, List, Optional + +from .transform_common import ( + _FEATURE_FLAG, + DEFAULT_MAX_OUTPUT_BYTES, + DEFAULT_MAX_TRANSFORMS_PER_REQUEST, + DEFAULT_TRANSFORM_TIMEOUT_SEC, + MAX_TRANSFORM_MODULE_SIZE_BYTES, + TransformLimits, + TransformRegistry, + TransformRegistryError, + TransformResult, + TransformStatus, + TrustedTransform, + get_transform_registry, + is_transforms_enabled, +) logger = logging.getLogger("ComfyUI-OpenClaw.services.constrained_transforms") -# --------------------------------------------------------------------------- -# Feature gate -# --------------------------------------------------------------------------- - -_FEATURE_FLAG = "OPENCLAW_ENABLE_TRANSFORMS" - - -def is_transforms_enabled() -> bool: - """Check if constrained transforms are enabled (default: OFF).""" - val = os.environ.get(_FEATURE_FLAG, "").strip().lower() - return val in ("1", "true", "yes", "on") - - -# --------------------------------------------------------------------------- -# Runtime limits -# --------------------------------------------------------------------------- - -DEFAULT_TRANSFORM_TIMEOUT_SEC = 5 -DEFAULT_MAX_OUTPUT_BYTES = 64 * 1024 # 64KB -DEFAULT_MAX_TRANSFORMS_PER_REQUEST = 5 -MAX_TRANSFORM_MODULE_SIZE_BYTES = 50 * 1024 # 50KB — prevent loading huge scripts - - -@dataclass -class TransformLimits: - """Runtime limits for transform execution.""" - - timeout_sec: float = DEFAULT_TRANSFORM_TIMEOUT_SEC - max_output_bytes: int = DEFAULT_MAX_OUTPUT_BYTES - max_transforms_per_request: int = DEFAULT_MAX_TRANSFORMS_PER_REQUEST - - @classmethod - def from_env(cls) -> "TransformLimits": - """Load limits from environment variables.""" - - def _env_int(key: str, default: int) -> int: - try: - return int(os.environ.get(key, str(default))) - except (ValueError, TypeError): - return default - - def _env_float(key: str, default: float) -> float: - try: - return float(os.environ.get(key, str(default))) - except (ValueError, TypeError): - return default - - return cls( - timeout_sec=_env_float( - "OPENCLAW_TRANSFORM_TIMEOUT", DEFAULT_TRANSFORM_TIMEOUT_SEC - ), - max_output_bytes=_env_int( - "OPENCLAW_TRANSFORM_MAX_OUTPUT", DEFAULT_MAX_OUTPUT_BYTES - ), - max_transforms_per_request=_env_int( - "OPENCLAW_TRANSFORM_MAX_PER_REQUEST", DEFAULT_MAX_TRANSFORMS_PER_REQUEST - ), - ) - - -# --------------------------------------------------------------------------- -# Transform result -# --------------------------------------------------------------------------- - - -class TransformStatus(str, Enum): - SUCCESS = "success" - ERROR = "error" - TIMEOUT = "timeout" - DENIED = "denied" - SKIPPED = "skipped" - - -@dataclass -class TransformResult: - """Result of a single transform execution.""" - - transform_id: str - status: str # TransformStatus.value - output: Optional[Dict[str, Any]] = None - error: str = "" - duration_ms: float = 0.0 - output_bytes: int = 0 - audit: Dict[str, Any] = field(default_factory=dict) - - def to_dict(self) -> Dict[str, Any]: - d: Dict[str, Any] = { - "transform_id": self.transform_id, - "status": self.status, - "duration_ms": round(self.duration_ms, 2), - "output_bytes": self.output_bytes, - } - if self.output is not None: - d["output"] = self.output - if self.error: - d["error"] = self.error - if self.audit: - d["audit"] = self.audit - return d - - -# --------------------------------------------------------------------------- -# Transform registry (trusted modules) -# --------------------------------------------------------------------------- - - -@dataclass -class TrustedTransform: - """A registered, integrity-pinned transform module.""" - - id: str - label: str - module_path: str # Absolute path to .py module - sha256: str # Integrity hash of the module file - description: str = "" - trusted_source: str = "" # Who published this transform - registered_at: float = 0.0 - - def to_dict(self) -> Dict[str, Any]: - return asdict(self) - - -class TransformRegistryError(Exception): - """Error in transform registry operations.""" - - pass - - -class TransformRegistry: - """ - Manages trusted transform modules with integrity pinning. - - Transforms can only be loaded from explicitly trusted directories. - Each module is pinned by its SHA256 hash at registration time. - """ - - def __init__(self, state_dir: str, trusted_dirs: Optional[List[str]] = None): - self._state_dir = state_dir - self._registry_dir = os.path.join(state_dir, "transforms") - self._index_path = os.path.join(self._registry_dir, "registry.json") - self._transforms: Dict[str, TrustedTransform] = {} - - # Trusted directories where transform modules can live - self._trusted_dirs: Set[str] = set() - if trusted_dirs: - for d in trusted_dirs: - resolved = str(Path(d).resolve()) - self._trusted_dirs.add(resolved) - - os.makedirs(self._registry_dir, exist_ok=True) - self._load() - - def _load(self) -> None: - """Load transform registry from disk.""" - if not os.path.exists(self._index_path): - self._transforms = {} - return - try: - with open(self._index_path, "r", encoding="utf-8") as f: - data = json.load(f) - self._transforms = {} - for tid, tdata in data.items(): - self._transforms[tid] = TrustedTransform(**tdata) - except Exception as e: - logger.error(f"Failed to load transform registry: {e}") - self._transforms = {} - - def _save(self) -> None: - """Persist transform registry to disk.""" - try: - data = {k: v.to_dict() for k, v in self._transforms.items()} - tmp_path = self._index_path + ".tmp" - with open(tmp_path, "w", encoding="utf-8") as f: - json.dump(data, f, indent=2) - f.write("\n") - os.replace(tmp_path, self._index_path) - except Exception as e: - logger.error(f"Failed to save transform registry: {e}") - - @staticmethod - def _compute_sha256(file_path: str) -> str: - """Compute SHA256 hash of a file.""" - h = hashlib.sha256() - with open(file_path, "rb") as f: - for chunk in iter(lambda: f.read(4096), b""): - h.update(chunk) - return h.hexdigest() - - def _is_in_trusted_dir(self, module_path: str) -> bool: - """Check if a module path is inside a trusted directory.""" - resolved = str(Path(module_path).resolve()) - for trusted in self._trusted_dirs: - if resolved.startswith(trusted + os.sep) or resolved == trusted: - return True - return False - - def register_transform( - self, - transform_id: str, - module_path: str, - *, - label: str = "", - description: str = "", - trusted_source: str = "", - ) -> TrustedTransform: - """ - Register a transform module with integrity pinning. - - The module must be in a trusted directory and within size limits. - """ - if not is_transforms_enabled(): - raise TransformRegistryError( - f"Transforms disabled. Set {_FEATURE_FLAG}=1 to enable." - ) - - abs_path = str(Path(module_path).resolve()) - - # Security: must be in trusted directory - if not self._is_in_trusted_dir(abs_path): - raise TransformRegistryError( - f"Module path is not in a trusted directory: {abs_path}" - ) - - if not os.path.isfile(abs_path): - raise TransformRegistryError(f"Module file not found: {abs_path}") - - # Size check - file_size = os.path.getsize(abs_path) - if file_size > MAX_TRANSFORM_MODULE_SIZE_BYTES: - raise TransformRegistryError( - f"Module exceeds size limit ({file_size} > {MAX_TRANSFORM_MODULE_SIZE_BYTES})" - ) - - # Must be a .py file - if not abs_path.endswith(".py"): - raise TransformRegistryError("Only .py modules are allowed as transforms") - - sha256 = self._compute_sha256(abs_path) - - transform = TrustedTransform( - id=transform_id, - label=label or transform_id, - module_path=abs_path, - sha256=sha256, - description=description, - trusted_source=trusted_source, - registered_at=time.time(), - ) - - self._transforms[transform_id] = transform - self._save() - logger.info(f"F42: Registered transform '{transform_id}' from {abs_path}") - return transform - - def unregister_transform(self, transform_id: str) -> bool: - """Remove a transform from the registry.""" - if not is_transforms_enabled(): - raise TransformRegistryError( - f"Transforms disabled. Set {_FEATURE_FLAG}=1 to enable." - ) - - if transform_id not in self._transforms: - return False - - del self._transforms[transform_id] - self._save() - logger.info(f"F42: Unregistered transform '{transform_id}'") - return True - - def get_transform(self, transform_id: str) -> Optional[TrustedTransform]: - """Get a registered transform by ID.""" - return self._transforms.get(transform_id) - - def list_transforms(self) -> List[TrustedTransform]: - """List all registered transforms.""" - return list(self._transforms.values()) - - def verify_integrity(self, transform_id: str) -> bool: - """Verify that a registered transform's file hasn't been modified.""" - transform = self._transforms.get(transform_id) - if not transform: - return False - - if not os.path.isfile(transform.module_path): - return False - - actual_hash = self._compute_sha256(transform.module_path) - return actual_hash == transform.sha256 +# IMPORTANT: +# Keep compatibility exports in this module even after refactor to +# `services.transform_common`; tests and downstream imports still reference +# `services.constrained_transforms` directly. +__all__ = [ + "_FEATURE_FLAG", + "DEFAULT_MAX_OUTPUT_BYTES", + "DEFAULT_MAX_TRANSFORMS_PER_REQUEST", + "DEFAULT_TRANSFORM_TIMEOUT_SEC", + "MAX_TRANSFORM_MODULE_SIZE_BYTES", + "TransformLimits", + "TransformRegistry", + "TransformRegistryError", + "TransformResult", + "TransformStatus", + "TrustedTransform", + "TransformExecutor", + "TransformTimeoutError", + "get_transform_executor", + "get_transform_registry", + "is_transforms_enabled", +] # --------------------------------------------------------------------------- @@ -536,47 +280,34 @@ class TransformExecutor: # Module-level convenience # --------------------------------------------------------------------------- -_registry: Optional[TransformRegistry] = None + _executor: Optional[TransformExecutor] = None -def get_transform_registry() -> TransformRegistry: - """Get or create the global transform registry.""" - global _registry - if _registry is None: - try: - from .state_dir import get_state_dir - - state_dir = get_state_dir() - except ImportError: - try: - from services.state_dir import get_state_dir - - state_dir = get_state_dir() - except ImportError: - state_dir = os.path.join( - os.path.dirname(os.path.dirname(__file__)), "data" - ) - - # Default trusted directory: pack-local transforms dir - pack_root = Path(__file__).resolve().parent.parent - trusted_dirs = [str(pack_root / "data" / "transforms")] - - # Allow additional trusted dirs from env - extra = os.environ.get("OPENCLAW_TRANSFORM_TRUSTED_DIRS", "") - if extra: - for d in extra.split(os.pathsep): - d = d.strip() - if d: - trusted_dirs.append(d) - - _registry = TransformRegistry(state_dir, trusted_dirs=trusted_dirs) - return _registry - - def get_transform_executor() -> TransformExecutor: - """Get or create the global transform executor.""" + """ + Get or create the global transform executor. + + If S35 isolation is enabled (default: True in this hardening wave), + returns a TransformProcessRunner instance. + """ global _executor if _executor is None: - _executor = TransformExecutor(get_transform_registry()) + # Check for process isolation flag (defaulting to on for S35) + # We can use the same enable flag, or a specific isolation one. + # Let's assume strict isolation is part of the enabling. + + # Local import to avoid circular dependency + try: + from .transform_runner import TransformProcessRunner + + registry = get_transform_registry() + # We treat TransformProcessRunner as compatible with TransformExecutor interface + _executor = TransformProcessRunner(registry) # type: ignore + except ImportError as e: + logger.warning( + f"S35: Could not import transform_runner ({e}), falling back to thread executor." + ) + _executor = TransformExecutor(get_transform_registry()) + return _executor diff --git a/services/csrf_protection.py b/services/csrf_protection.py index 0047e8e..402ac33 100644 --- a/services/csrf_protection.py +++ b/services/csrf_protection.py @@ -1,3 +1,5 @@ +from __future__ import annotations + """ S26+: Same-Origin CSRF Protection for Localhost Convenience Mode @@ -7,16 +9,38 @@ when no admin token is configured (convenience mode). Purpose: Prevent cross-origin requests from abusing localhost-only admin endpoints. """ +import json import logging import os from typing import Optional -from aiohttp import web +try: + from aiohttp import web # type: ignore +except ModuleNotFoundError: # pragma: no cover - CI/minimal env path + web = None # type: ignore + +# IMPORTANT: +# Keep this module importable even when aiohttp is unavailable (CI/minimal env). +# Tests import `is_same_origin_request` directly; a hard import error here causes +# unrelated auth-hardening suites to fail before skip guards can apply. + + +class _CompatResponse: + """Minimal fallback response when aiohttp is unavailable.""" + + def __init__(self, *, status: int, body: bytes, content_type: str): + self.status = status + self.body = body + self.text = body.decode("utf-8", errors="replace") + self.content_type = content_type + + +def _json_response(payload: dict, *, status: int): + if web is not None: + return web.json_response(payload, status=status) + body = json.dumps(payload, ensure_ascii=False).encode("utf-8") + return _CompatResponse(status=status, body=body, content_type="application/json") -# NOTE FOR CALLERS: -# This module has an import-time aiohttp dependency. If a caller must remain -# importable in minimal test/CI environments without aiohttp, guard the import -# at the caller boundary (see `api/config.py` for the required pattern). logger = logging.getLogger("ComfyUI-OpenClaw.services.csrf_protection") @@ -117,7 +141,7 @@ def require_same_origin_if_no_token( logger.warning( f"S26+: CSRF protection denied cross-origin request to {request.path}" ) - return web.json_response( + return _json_response( { "ok": False, "error": "csrf_protection", diff --git a/services/integrity.py b/services/integrity.py new file mode 100644 index 0000000..3f03688 --- /dev/null +++ b/services/integrity.py @@ -0,0 +1,136 @@ +""" +R77 Integrity Envelopes. + +Provides canonical serialization and integrity verification for persisted state. +""" + +import hashlib +import json +import logging +import os +import shutil +import tempfile +from dataclasses import asdict, dataclass +from typing import Any, Dict, Optional, Union + +logger = logging.getLogger("ComfyUI-OpenClaw.services.integrity") + + +@dataclass +class IntegrityEnvelope: + """ + Wrapper for persisted data with integrity metadata. + """ + + version: int + data: Dict[str, Any] + hash: str # SHA256 of canonical(data) + algo: str = "sha256" + meta: Optional[Dict[str, Any]] = None + + +class IntegrityError(Exception): + """Raised when integrity verification fails.""" + + pass + + +def canonical_dumps(data: Any) -> bytes: + """ + Serialize data to canonical JSON (sorted keys, no whitespace). + """ + return json.dumps(data, sort_keys=True, separators=(",", ":")).encode("utf-8") + + +def calculate_hash(data: Any, algo: str = "sha256") -> str: + """ + Calculate hash of canonicalized data. + """ + if algo != "sha256": + raise ValueError(f"Unsupported hash algorithm: {algo}") + + payload = canonical_dumps(data) + return hashlib.sha256(payload).hexdigest() + + +def save_verified(path: str, data: Dict[str, Any], version: int = 1) -> None: + """ + Save data wrapped in an integrity envelope. + Atomic write. + """ + data_hash = calculate_hash(data) + envelope = IntegrityEnvelope( + version=version, data=data, hash=data_hash, algo="sha256" + ) + + # Write to temp string first to Ensure serialization works + try: + content = json.dumps(asdict(envelope), indent=2) + except Exception as e: + logger.error(f"Failed to serialize integrity envelope for {path}: {e}") + raise + + # Atomic write + dir_name = os.path.dirname(os.path.abspath(path)) + os.makedirs(dir_name, exist_ok=True) + + fd, tmp_path = tempfile.mkstemp(dir=dir_name, text=True) + try: + with os.fdopen(fd, "w", encoding="utf-8") as f: + f.write(content) + f.flush() + os.fsync(fd) + + # Renaissance-style atomic rename + os.replace(tmp_path, path) + except Exception as e: + logger.error(f"Failed to save verified file {path}: {e}") + if os.path.exists(tmp_path): + os.remove(tmp_path) + raise + + +def load_verified( + path: str, expected_version: int = 1, migrate: bool = True +) -> Dict[str, Any]: + """ + Load data from an integrity envelope. + + If `migrate` is True and the file is valid legacy JSON (no envelope), + it returns the data as-is (caller should save back to upgrade). + + Raises IntegrityError if hash mismatch or malformed. + """ + if not os.path.exists(path): + raise FileNotFoundError(f"File not found: {path}") + + try: + with open(path, "r", encoding="utf-8") as f: + raw = json.load(f) + except json.JSONDecodeError as e: + raise IntegrityError(f"Corrupt JSON file {path}: {e}") + + # Check if it's an envelope + if isinstance(raw, dict) and "hash" in raw and "data" in raw and "version" in raw: + # Verify integrity + stored_hash = raw["hash"] + stored_data = raw["data"] + + computed_hash = calculate_hash(stored_data) + if computed_hash != stored_hash: + raise IntegrityError(f"Integrity check failed for {path} (hash mismatch)") + + # Verify version if needed + # We can implement version migration logic here if multiple envelope versions exist + + return stored_data + + # Legacy Fallback + if migrate: + logger.info( + f"R77: Loaded legacy file {path}, integrity check skipped (pending migration)." + ) + # For legacy files, we assume the whole content is the data. + return raw + + raise IntegrityError(f"File {path} is not a valid integrity envelope") diff --git a/services/runtime_config.py b/services/runtime_config.py index 0e120f9..6809b27 100644 --- a/services/runtime_config.py +++ b/services/runtime_config.py @@ -145,6 +145,12 @@ SCHEDULER_ENV_MAPPINGS = { "skip_missed_intervals": ("OPENCLAW_SCHEDULER_SKIP_MISSED", ""), } +# IMPORTANT: +# Keep effective-config merge order deterministic. +# Using a set iteration here makes legacy warning assertions flaky because the +# first env key read can vary per process/hash seed. +LLM_KEY_ORDER = tuple(ENV_MAPPINGS.keys()) + def _clamp(value: int, min_val: int, max_val: int) -> int: """Clamp an integer to a range.""" @@ -271,7 +277,11 @@ def get_effective_config() -> Tuple[Dict[str, Any], Dict[str, str]]: effective = {} sources = {} - for key in ALLOWED_LLM_KEYS: + ordered_keys = list(LLM_KEY_ORDER) + [ + k for k in sorted(ALLOWED_LLM_KEYS) if k not in ENV_MAPPINGS + ] + + for key in ordered_keys: # 1. Check ENV override env_val = _get_env_value(key) if env_val is not None: diff --git a/services/transform_common.py b/services/transform_common.py new file mode 100644 index 0000000..9583296 --- /dev/null +++ b/services/transform_common.py @@ -0,0 +1,348 @@ +""" +Common types and registry for constrained transforms (S35/F42). +Refactored to avoid circular imports. +""" + +from __future__ import annotations + +import hashlib +import json +import logging +import os +import time +from dataclasses import asdict, dataclass, field +from enum import Enum +from pathlib import Path +from typing import Any, Dict, List, Optional, Set + +logger = logging.getLogger("ComfyUI-OpenClaw.services.transform_common") + +# --------------------------------------------------------------------------- +# Feature gate +# --------------------------------------------------------------------------- + +_FEATURE_FLAG = "OPENCLAW_ENABLE_TRANSFORMS" + + +def is_transforms_enabled() -> bool: + """Check if constrained transforms are enabled (default: OFF).""" + val = os.environ.get(_FEATURE_FLAG, "").strip().lower() + return val in ("1", "true", "yes", "on") + + +# --------------------------------------------------------------------------- +# Runtime limits +# --------------------------------------------------------------------------- + +DEFAULT_TRANSFORM_TIMEOUT_SEC = 5 +DEFAULT_MAX_OUTPUT_BYTES = 64 * 1024 # 64KB +DEFAULT_MAX_TRANSFORMS_PER_REQUEST = 5 +MAX_TRANSFORM_MODULE_SIZE_BYTES = 50 * 1024 # 50KB — prevent loading huge scripts + + +@dataclass +class TransformLimits: + """Runtime limits for transform execution.""" + + timeout_sec: float = DEFAULT_TRANSFORM_TIMEOUT_SEC + max_output_bytes: int = DEFAULT_MAX_OUTPUT_BYTES + max_transforms_per_request: int = DEFAULT_MAX_TRANSFORMS_PER_REQUEST + + @classmethod + def from_env(cls) -> "TransformLimits": + """Load limits from environment variables.""" + + def _env_int(key: str, default: int) -> int: + try: + return int(os.environ.get(key, str(default))) + except (ValueError, TypeError): + return default + + def _env_float(key: str, default: float) -> float: + try: + return float(os.environ.get(key, str(default))) + except (ValueError, TypeError): + return default + + return cls( + timeout_sec=_env_float( + "OPENCLAW_TRANSFORM_TIMEOUT", DEFAULT_TRANSFORM_TIMEOUT_SEC + ), + max_output_bytes=_env_int( + "OPENCLAW_TRANSFORM_MAX_OUTPUT", DEFAULT_MAX_OUTPUT_BYTES + ), + max_transforms_per_request=_env_int( + "OPENCLAW_TRANSFORM_MAX_PER_REQUEST", DEFAULT_MAX_TRANSFORMS_PER_REQUEST + ), + ) + + +# --------------------------------------------------------------------------- +# Transform result +# --------------------------------------------------------------------------- + + +class TransformStatus(str, Enum): + SUCCESS = "success" + ERROR = "error" + TIMEOUT = "timeout" + DENIED = "denied" + SKIPPED = "skipped" + + +@dataclass +class TransformResult: + """Result of a single transform execution.""" + + transform_id: str + status: str # TransformStatus.value + output: Optional[Dict[str, Any]] = None + error: str = "" + duration_ms: float = 0.0 + output_bytes: int = 0 + audit: Dict[str, Any] = field(default_factory=dict) + + def to_dict(self) -> Dict[str, Any]: + d: Dict[str, Any] = { + "transform_id": self.transform_id, + "status": self.status, + "duration_ms": round(self.duration_ms, 2), + "output_bytes": self.output_bytes, + } + if self.output is not None: + d["output"] = self.output + if self.error: + d["error"] = self.error + if self.audit: + d["audit"] = self.audit + return d + + +# --------------------------------------------------------------------------- +# Transform registry (trusted modules) +# --------------------------------------------------------------------------- + + +@dataclass +class TrustedTransform: + """A registered, integrity-pinned transform module.""" + + id: str + label: str + module_path: str # Absolute path to .py module + sha256: str # Integrity hash of the module file + description: str = "" + trusted_source: str = "" # Who published this transform + registered_at: float = 0.0 + + def to_dict(self) -> Dict[str, Any]: + return asdict(self) + + +class TransformRegistryError(Exception): + """Error in transform registry operations.""" + + pass + + +class TransformRegistry: + """ + Manages trusted transform modules with integrity pinning. + + Transforms can only be loaded from explicitly trusted directories. + Each module is pinned by its SHA256 hash at registration time. + """ + + def __init__(self, state_dir: str, trusted_dirs: Optional[List[str]] = None): + self._state_dir = state_dir + self._registry_dir = os.path.join(state_dir, "transforms") + self._index_path = os.path.join(self._registry_dir, "registry.json") + self._transforms: Dict[str, TrustedTransform] = {} + + # Trusted directories where transform modules can live + self._trusted_dirs: Set[str] = set() + if trusted_dirs: + for d in trusted_dirs: + resolved = str(Path(d).resolve()) + self._trusted_dirs.add(resolved) + + os.makedirs(self._registry_dir, exist_ok=True) + self._load() + + def _load(self) -> None: + """Load transform registry from disk.""" + if not os.path.exists(self._index_path): + self._transforms = {} + return + try: + with open(self._index_path, "r", encoding="utf-8") as f: + data = json.load(f) + self._transforms = {} + for tid, tdata in data.items(): + self._transforms[tid] = TrustedTransform(**tdata) + except Exception as e: + logger.error(f"Failed to load transform registry: {e}") + self._transforms = {} + + def _save(self) -> None: + """Persist transform registry to disk.""" + try: + data = {k: v.to_dict() for k, v in self._transforms.items()} + tmp_path = self._index_path + ".tmp" + with open(tmp_path, "w", encoding="utf-8") as f: + json.dump(data, f, indent=2) + f.write("\n") + os.replace(tmp_path, self._index_path) + except Exception as e: + logger.error(f"Failed to save transform registry: {e}") + + @staticmethod + def _compute_sha256(file_path: str) -> str: + """Compute SHA256 hash of a file.""" + h = hashlib.sha256() + with open(file_path, "rb") as f: + for chunk in iter(lambda: f.read(4096), b""): + h.update(chunk) + return h.hexdigest() + + def _is_in_trusted_dir(self, module_path: str) -> bool: + """Check if a module path is inside a trusted directory.""" + resolved = str(Path(module_path).resolve()) + for trusted in self._trusted_dirs: + if resolved.startswith(trusted + os.sep) or resolved == trusted: + return True + return False + + def register_transform( + self, + transform_id: str, + module_path: str, + *, + label: str = "", + description: str = "", + trusted_source: str = "", + ) -> TrustedTransform: + """ + Register a transform module with integrity pinning. + + The module must be in a trusted directory and within size limits. + """ + if not is_transforms_enabled(): + raise TransformRegistryError( + f"Transforms disabled. Set {_FEATURE_FLAG}=1 to enable." + ) + + abs_path = str(Path(module_path).resolve()) + + # Security: must be in trusted directory + if not self._is_in_trusted_dir(abs_path): + raise TransformRegistryError( + f"Module path is not in a trusted directory: {abs_path}" + ) + + if not os.path.isfile(abs_path): + raise TransformRegistryError(f"Module file not found: {abs_path}") + + # Size check + file_size = os.path.getsize(abs_path) + if file_size > MAX_TRANSFORM_MODULE_SIZE_BYTES: + raise TransformRegistryError( + f"Module exceeds size limit ({file_size} > {MAX_TRANSFORM_MODULE_SIZE_BYTES})" + ) + + # Must be a .py file + if not abs_path.endswith(".py"): + raise TransformRegistryError("Only .py modules are allowed as transforms") + + sha256 = self._compute_sha256(abs_path) + + transform = TrustedTransform( + id=transform_id, + label=label or transform_id, + module_path=abs_path, + sha256=sha256, + description=description, + trusted_source=trusted_source, + registered_at=time.time(), + ) + + self._transforms[transform_id] = transform + self._save() + logger.info(f"F42: Registered transform '{transform_id}' from {abs_path}") + return transform + + def unregister_transform(self, transform_id: str) -> bool: + """Remove a transform from the registry.""" + if not is_transforms_enabled(): + raise TransformRegistryError( + f"Transforms disabled. Set {_FEATURE_FLAG}=1 to enable." + ) + + if transform_id not in self._transforms: + return False + + del self._transforms[transform_id] + self._save() + logger.info(f"F42: Unregistered transform '{transform_id}'") + return True + + def get_transform(self, transform_id: str) -> Optional[TrustedTransform]: + """Get a registered transform by ID.""" + return self._transforms.get(transform_id) + + def list_transforms(self) -> List[TrustedTransform]: + """List all registered transforms.""" + return list(self._transforms.values()) + + def verify_integrity(self, transform_id: str) -> bool: + """Verify that a registered transform's file hasn't been modified.""" + transform = self._transforms.get(transform_id) + if not transform: + return False + + if not os.path.isfile(transform.module_path): + return False + + actual_hash = self._compute_sha256(transform.module_path) + return actual_hash == transform.sha256 + + +# --------------------------------------------------------------------------- +# Module-level convenience +# --------------------------------------------------------------------------- + +_registry: Optional[TransformRegistry] = None + + +def get_transform_registry() -> TransformRegistry: + """Get or create the global transform registry.""" + global _registry + if _registry is None: + try: + from .state_dir import get_state_dir + + state_dir = get_state_dir() + except ImportError: + try: + from services.state_dir import get_state_dir + + state_dir = get_state_dir() + except ImportError: + state_dir = os.path.join( + os.path.dirname(os.path.dirname(__file__)), "data" + ) + + # Default trusted directory: pack-local transforms dir + pack_root = Path(__file__).resolve().parent.parent + trusted_dirs = [str(pack_root / "data" / "transforms")] + + # Allow additional trusted dirs from env + extra = os.environ.get("OPENCLAW_TRANSFORM_TRUSTED_DIRS", "") + if extra: + for d in extra.split(os.pathsep): + d = d.strip() + if d: + trusted_dirs.append(d) + + _registry = TransformRegistry(state_dir, trusted_dirs=trusted_dirs) + return _registry diff --git a/services/transform_runner.py b/services/transform_runner.py new file mode 100644 index 0000000..162db94 --- /dev/null +++ b/services/transform_runner.py @@ -0,0 +1,183 @@ +""" +S35 Transform Isolation Runner. + +Executes transforms in a separate process via `services.transform_worker`. +""" + +import json +import logging +import os +import subprocess +import sys +import time +from typing import Any, Dict, Optional + +from .transform_common import ( + TransformLimits, + TransformRegistry, + TransformResult, + TransformStatus, +) + +logger = logging.getLogger("ComfyUI-OpenClaw.services.transform_runner") + + +class TransformProcessRunner: + """ + Executes transforms in an isolated subprocess. + """ + + def __init__( + self, + registry: TransformRegistry, + limits: Optional[TransformLimits] = None, + ): + self._registry = registry + self._limits = limits or TransformLimits.from_env() + + def execute_transform( + self, + transform_id: str, + input_data: Dict[str, Any], + *, + trace_id: str = "", + ) -> TransformResult: + """ + Execute a single registered transform in a subprocess. + """ + transform = self._registry.get_transform(transform_id) + if not transform: + return TransformResult( + transform_id=transform_id, + status=TransformStatus.ERROR.value, + error=f"Transform '{transform_id}' not found in registry", + ) + + # Integrity Check (R77 pre-check) + if not self._registry.verify_integrity(transform_id): + return TransformResult( + transform_id=transform_id, + status=TransformStatus.DENIED.value, + error="Integrity verification failed — module modified", + audit={"reason": "integrity_check_failed", "trace_id": trace_id}, + ) + + start_time = time.monotonic() + + # Prepare Worker Command + worker_script = os.path.join(os.path.dirname(__file__), "transform_worker.py") + cmd = [sys.executable, worker_script, transform.module_path] + + # Prepare Input + payload = {"input": input_data, "context": {"trace_id": trace_id}} + input_json = json.dumps(payload) + + try: + # capability-deny: no environment variable inheritance by default? + # Or minimal env. + env = os.environ.copy() + # Remove sensitive vars if needed? + # S34 Obs: redact env in logs, but here process sees env. + # Best practice: clear sensitive vars. + for key in list(env.keys()): + if "TOKEN" in key or "SECRET" in key or "KEY" in key: + del env[key] + + # Subprocess Run + proc = subprocess.run( + cmd, + input=input_json, + capture_output=True, + text=True, + timeout=self._limits.timeout_sec, + env=env, + check=False, # We handle return codes + ) + + # Execution Time + elapsed_ms = (time.monotonic() - start_time) * 1000 + + # Handle Return Code + if proc.returncode != 0: + # Script crashed or printed error to stdout/stderr + # Try to parse stdout error first + error_msg = proc.stderr.strip() or "Process crashed with unknown error" + try: + out_json = json.loads(proc.stdout) + if out_json.get("status") == "error": + error_msg = out_json.get("error", error_msg) + except Exception: + pass + + return TransformResult( + transform_id=transform_id, + status=TransformStatus.ERROR.value, + error=f"Worker process failed (exit {proc.returncode}): {error_msg}", + duration_ms=elapsed_ms, + audit={"exit_code": proc.returncode, "trace_id": trace_id}, + ) + + # Parse Output + try: + result_json = json.loads(proc.stdout) + except json.JSONDecodeError: + return TransformResult( + transform_id=transform_id, + status=TransformStatus.ERROR.value, + error="Worker returned invalid JSON output", + duration_ms=elapsed_ms, + audit={"raw_stdout": proc.stdout[:1000], "trace_id": trace_id}, + ) + + if result_json.get("status") == "error": + return TransformResult( + transform_id=transform_id, + status=TransformStatus.ERROR.value, + error=result_json.get("error", "Unknown worker error"), + duration_ms=elapsed_ms, + audit={ + "trace_id": trace_id, + "traceback": result_json.get("traceback"), + }, + ) + + output_data = result_json.get("output", {}) + output_bytes = len(json.dumps(output_data).encode("utf-8")) + + if output_bytes > self._limits.max_output_bytes: + return TransformResult( + transform_id=transform_id, + status=TransformStatus.ERROR.value, + error=f"Output size limit exceeded ({output_bytes} > {self._limits.max_output_bytes})", + duration_ms=elapsed_ms, + output_bytes=output_bytes, + audit={"trace_id": trace_id}, + ) + + return TransformResult( + transform_id=transform_id, + status=TransformStatus.SUCCESS.value, + output=output_data, + duration_ms=elapsed_ms, + output_bytes=output_bytes, + audit={"trace_id": trace_id, "isolation": "process"}, + ) + + except subprocess.TimeoutExpired: + elapsed_ms = (time.monotonic() - start_time) * 1000 + return TransformResult( + transform_id=transform_id, + status=TransformStatus.TIMEOUT.value, + error=f"Transform timeout exceeded ({self._limits.timeout_sec}s)", + duration_ms=elapsed_ms, + audit={"trace_id": trace_id, "timeout": True}, + ) + except Exception as e: + elapsed_ms = (time.monotonic() - start_time) * 1000 + return TransformResult( + transform_id=transform_id, + status=TransformStatus.ERROR.value, + error=f"Runner exception: {str(e)}", + duration_ms=elapsed_ms, + audit={"trace_id": trace_id}, + ) diff --git a/services/transform_worker.py b/services/transform_worker.py new file mode 100644 index 0000000..4f40ae5 --- /dev/null +++ b/services/transform_worker.py @@ -0,0 +1,102 @@ +""" +S35 Transform Isolation Worker. + +This script runs in a separate process to execute a transform module. +It reads input JSON from stdin and writes output JSON to stdout. + +Usage: + python -m services.transform_worker + +Protocol: + Input (stdin): JSON object {"input": {...}, "context": {...}} + Output (stdout): JSON object {"status": "success", "output": {...}} or {"status": "error", "error": "..."} + Exit Code: 0 on success/handled error, non-zero on crash. +""" + +import argparse +import importlib.util +import json +import os +import socket +import sys +import traceback +from typing import Any, Dict + +# S35: Capability Deny-by-Default +# We can't easily drop OS privileges in a cross-platform way without deps, +# but being in a separate process isolates memory and crashes. +# We could monkeypatch network libs here to enforce "no network". + + +def _deny_network(*args, **kwargs): + raise RuntimeError("Network access denied by S35 transform isolation policy") + + +# Monkeypatch socket to deny network access +socket.socket = _deny_network +socket.create_connection = _deny_network +# TODO: Monkeypatch http.client, urllib, requests if present? +# Standard library socket blocks most. + + +def load_module(module_path: str): + """Load the transform module from path.""" + if not os.path.exists(module_path): + raise FileNotFoundError(f"Module not found: {module_path}") + + spec = importlib.util.spec_from_file_location("transform_module", module_path) + if not spec or not spec.loader: + raise ImportError(f"Could not load spec for {module_path}") + + module = importlib.util.module_from_spec(spec) + sys.modules["transform_module"] = module + spec.loader.exec_module(module) + return module + + +def main(): + parser = argparse.ArgumentParser() + parser.add_argument("module_path", help="Absolute path to the transform module") + args = parser.parse_args() + + # Read input payload + try: + input_raw = sys.stdin.read() + if not input_raw: + raise ValueError("Empty input on stdin") + payload = json.loads(input_raw) + data = payload.get("input", {}) + except Exception as e: + response = {"status": "error", "error": f"Failed to read input: {e}"} + print(json.dumps(response)) + sys.exit(1) + + # Execute transform + try: + module = load_module(args.module_path) + + if not hasattr(module, "transform"): + raise AttributeError("Module missing 'transform' function") + + func = getattr(module, "transform") + if not callable(func): + raise TypeError("'transform' is not callable") + + # Run + result = func(data) + + # Validate result (JSON serializable?) + # We try to dump it. If it fails, that's an error. + output_payload = {"status": "success", "output": result} + print(json.dumps(output_payload, default=str)) + + except Exception as e: + # Capture traceback for diagnostics + tb = traceback.format_exc() + response = {"status": "error", "error": str(e), "traceback": tb} + print(json.dumps(response)) + sys.exit(0) # Exit 0 because we handled the error gracefully + + +if __name__ == "__main__": + main() diff --git a/tests/TEST_SOP.md b/tests/TEST_SOP.md index 6b76c2e..a096964 100644 --- a/tests/TEST_SOP.md +++ b/tests/TEST_SOP.md @@ -11,6 +11,7 @@ Every implementation plan must include the **full test validation procedure** in - Python 3.10+ (CI uses 3.10/3.11) - Node.js 18+ (CI uses 20) - `pre-commit` installed: `python -m pip install pre-commit` +- Backend test deps available in the same interpreter (`numpy`, `pillow`, `aiohttp`) - Frontend deps installed: `npm install` ## Environment Sanity (Required Guardrails) @@ -18,11 +19,13 @@ Every implementation plan must include the **full test validation procedure** in - **Python interpreter must be consistent** for all test commands. - Verify: `python -c "import sys; print(sys.executable)"` - If you use conda or venv, ensure the same interpreter runs unit tests and connector tests. -- **Project venv recommended**: use `.venv` when possible to avoid mixed dependencies. - - Create: `python -m venv .venv` - - Activate (bash): `source .venv/bin/activate` +- **Project venv recommended**: use an OS-specific local venv to avoid mixed dependencies. + - Linux/WSL recommended path: `.venv-wsl` (especially when Windows also uses `.venv` in the same repo) + - Other environments: `.venv` + - Create: `python -m venv .venv-wsl` (WSL) or `python -m venv .venv` + - Activate (bash): `source .venv-wsl/bin/activate` (or `.venv/bin/activate`) - Activate (pwsh): `.\.venv\Scripts\Activate.ps1` - - If tests fail due to missing deps in CI parity, **rerun in `.venv` and record that in the implementation record**. + - If tests fail due to missing deps in CI parity, rerun in the project venv used by scripts and record that in the implementation record. - **Node version must be 18+** before E2E: - Verify: `node -v` - If mismatch in WSL, use the Node 18 path specified below. @@ -121,8 +124,9 @@ Use these checks before assuming the hook runner is broken: ### Optional: One-Command Full Test Scripts (Fastest) Use these if you want a single command that runs **all required steps** (detect-secrets, pre-commit, unit tests, E2E). These scripts also handle the most common environment issues (Windows cache locks, Black cache, Node 18). -Both scripts enforce a project-local `.venv` and will bootstrap missing test tooling (`pre-commit`, and `aiohttp` where needed for imports). -If `.venv` exists but is invalid for the current OS (for example created in WSL then reused in Windows), rerun via the script so it can recreate the environment. +Scripts enforce a project-local venv and will bootstrap missing test tooling (`pre-commit`, and `aiohttp` where needed for imports). +On WSL, scripts prefer `.venv-wsl`; on Windows they use `.venv`. +If the selected venv exists but is invalid for the current OS/interpreter, rerun via the script so it can recreate that venv. Linux script includes an explicit offline fail-fast guard: if dependency bootstrap fails (for example `aiohttp` / `pre-commit` install), it stops with remediation hints instead of continuing with partial state. - Linux/WSL: @@ -144,6 +148,16 @@ Then every `git push` will run: bash scripts/pre_push_checks.sh ``` +`scripts/pre_push_checks.sh` is the CI-parity guard and must include all 4 stages: +1) `detect-secrets` +2) all `pre-commit` hooks +3) backend unit tests (`scripts/run_unittests.py --pattern "test_*.py"`) +4) frontend E2E (`npm test`) + +IMPORTANT: +- Do not remove stage (3). If pre-push skips backend unit tests, local pushes can pass while GitHub CI fails later. +- Keep dependency bootstrap in this script aligned with `.github/workflows/ci.yml` unit-test dependencies. + 1) Detect Secrets (baseline-based) ```bash diff --git a/tests/test_r77_integrity.py b/tests/test_r77_integrity.py new file mode 100644 index 0000000..9212675 --- /dev/null +++ b/tests/test_r77_integrity.py @@ -0,0 +1,107 @@ +""" +R77 Integrity Tests. +""" + +import json +import os +import shutil +import tempfile +import unittest + +from services.integrity import ( + IntegrityEnvelope, + IntegrityError, + calculate_hash, + canonical_dumps, + load_verified, + save_verified, +) + + +class TestR77Integrity(unittest.TestCase): + def setUp(self): + self.test_dir = tempfile.mkdtemp() + + def tearDown(self): + shutil.rmtree(self.test_dir) + + def test_canonical_serialization(self): + """Test that JSON is canonicalized (sorted keys, no whitespace).""" + data1 = {"b": 2, "a": 1} + data2 = {"a": 1, "b": 2} + + c1 = canonical_dumps(data1) + c2 = canonical_dumps(data2) + + self.assertEqual(c1, c2) + # Check format: {"a":1,"b":2} + self.assertEqual(c1, b'{"a":1,"b":2}') + + def test_hashing(self): + """Test hash consistency.""" + data = {"foo": "bar"} + h1 = calculate_hash(data) + h2 = calculate_hash(data) + self.assertEqual(h1, h2) + self.assertTrue(len(h1) == 64) # SHA256 hex digest + + def test_save_and_load_verified(self): + """Test saving and loading with integrity envelope.""" + path = os.path.join(self.test_dir, "test.json") + data = {"key": "value", "list": [1, 2, 3]} + + save_verified(path, data) + + # Verify file structure on disk + with open(path, "r") as f: + envelope = json.load(f) + + self.assertIn("version", envelope) + self.assertIn("hash", envelope) + self.assertIn("data", envelope) + self.assertEqual(envelope["data"], data) + + # Load back + loaded = load_verified(path) + self.assertEqual(loaded, data) + + def test_tamper_detection(self): + """Test that modification of data voids the integrity check.""" + path = os.path.join(self.test_dir, "tampered.json") + data = {"secret": "123"} + save_verified(path, data) + + # Tamper with the file + with open(path, "r") as f: + envelope = json.load(f) + + envelope["data"]["secret"] = "666" # Evil modification + # Hash is NOT updated + + with open(path, "w") as f: + json.dump(envelope, f) + + # Load should fail + with self.assertRaises(IntegrityError): + load_verified(path) + + def test_legacy_migration(self): + """Test verifying legacy (non-envelope) files.""" + path = os.path.join(self.test_dir, "legacy.json") + data = {"old": "data"} + + # Save as raw JSON + with open(path, "w") as f: + json.dump(data, f) + + # load_verified with migrate=True (default) should succeed + loaded = load_verified(path, migrate=True) + self.assertEqual(loaded, data) + + # load_verified with migrate=False should fail + with self.assertRaises(IntegrityError): + load_verified(path, migrate=False) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_runtime_config.py b/tests/test_runtime_config.py index d8417c1..0c4f31f 100644 --- a/tests/test_runtime_config.py +++ b/tests/test_runtime_config.py @@ -43,6 +43,20 @@ class TestRuntimeConfig(unittest.TestCase): patcher.start() self.addCleanup(patcher.stop) + # IMPORTANT: + # runtime_config._get_env_value caches legacy-warning emission in a function + # attribute (`_warned_legacy`). Without resetting it here, test order can + # suppress expected warning logs and produce false negatives in CI. + try: + for mod_name, mod in list(sys.modules.items()): + if not mod_name.endswith("runtime_config"): + continue + fn = getattr(mod, "_get_env_value", None) + if callable(fn): + setattr(fn, "_warned_legacy", set()) + except Exception: + pass + for key in [ "MOLTBOT_LLM_PROVIDER", "MOLTBOT_LLM_MODEL", diff --git a/tests/test_s33r76r78_auth_hardening.py b/tests/test_s33r76r78_auth_hardening.py index 0555da2..074fe3f 100644 --- a/tests/test_s33r76r78_auth_hardening.py +++ b/tests/test_s33r76r78_auth_hardening.py @@ -25,9 +25,16 @@ from services.csrf_protection import is_same_origin_request # classified here, otherwise tests fail. AUTH_CLASS_BY_ROUTE = { ("GET", "/health"): "public-safe", - ("GET", "/logs/tail"): "observability", + # IMPORTANT: + # `/logs/tail` was hardened to admin-only because log payload can expose + # high-sensitivity prompt/runtime context (S34). Keep this auth class in + # sync with api/logs_tail.py to avoid accidental privilege regression. + ("GET", "/logs/tail"): "admin", ("GET", "/jobs"): "public-safe", - ("GET", "/trace/{prompt_id}"): "observability", + # IMPORTANT: + # Trace endpoint now returns high-sensitivity execution context and is + # intentionally admin-only (S34). Keep as admin to prevent data leakage. + ("GET", "/trace/{prompt_id}"): "admin", ("POST", "/webhook"): "webhook-auth", ("POST", "/webhook/submit"): "webhook-auth", ("POST", "/webhook/validate"): "webhook-auth", diff --git a/tests/test_s34_observability_tiers.py b/tests/test_s34_observability_tiers.py new file mode 100644 index 0000000..54415bd --- /dev/null +++ b/tests/test_s34_observability_tiers.py @@ -0,0 +1,79 @@ +""" +S34 Observability Tiers Tests. +""" + +import os +import unittest +from unittest.mock import MagicMock, patch + +from services.access_control import require_admin_token, require_observability_access + + +# Mock helpers +class MockRequest: + def __init__(self, headers=None, remote="1.2.3.4"): + self.headers = headers or {} + self.remote = remote + self.match_info = {} + self.query = {} + + +class TestS34ObservabilityTiers(unittest.TestCase): + def setUp(self): + self.obs_token = "obs-secret" + self.admin_token = "admin-secret" + os.environ["OPENCLAW_OBSERVABILITY_TOKEN"] = self.obs_token + os.environ["OPENCLAW_ADMIN_TOKEN"] = self.admin_token + + def tearDown(self): + if "OPENCLAW_OBSERVABILITY_TOKEN" in os.environ: + del os.environ["OPENCLAW_OBSERVABILITY_TOKEN"] + if "OPENCLAW_ADMIN_TOKEN" in os.environ: + del os.environ["OPENCLAW_ADMIN_TOKEN"] + + def test_low_sensitivity_access(self): + """Health/Config should allow Obs Token.""" + req = MockRequest(headers={"X-OpenClaw-Obs-Token": self.obs_token}) + allowed, _ = require_observability_access(req) + self.assertTrue(allowed, "Obs token should allow low sensitivity access") + + def test_high_sensitivity_denial(self): + """Trace/Log should DENY Obs Token (require Admin).""" + # Admin check with ONLY obs token should fail + req = MockRequest(headers={"X-OpenClaw-Obs-Token": self.obs_token}) + allowed, _ = require_admin_token(req) + self.assertFalse(allowed, "Obs token should NOT pass Admin check") + + def test_high_sensitivity_allow(self): + """Trace/Log should ALLOW Admin Token.""" + req = MockRequest(headers={"X-Moltbot-Admin-Token": self.admin_token}) + allowed, _ = require_admin_token(req) + self.assertTrue(allowed, "Admin token should pass Admin check") + + @patch("api.routes.trace_store") + @patch("api.routes.require_admin_token") + @patch("api.routes.web") + def test_trace_handler_tier_enforcement( + self, mock_web, mock_require_admin, mock_trace_store + ): + """Verify API handler calls the right check.""" + import asyncio + + from api.routes import trace_handler + + # Setup + mock_require_admin.return_value = (False, "Denied") + req = MockRequest() + + async def run_test(): + await trace_handler(req) + + # Execute + asyncio.run(run_test()) + + # Verify + mock_require_admin.assert_called_once() + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_s35_transform_isolation.py b/tests/test_s35_transform_isolation.py new file mode 100644 index 0000000..20414d3 --- /dev/null +++ b/tests/test_s35_transform_isolation.py @@ -0,0 +1,135 @@ +""" +S35 Transform Isolation Tests. +""" + +import json +import os +import sys +import unittest +from unittest.mock import MagicMock, patch + +# Add project root to path +sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))) + +from services.transform_common import ( + TransformLimits, + TransformRegistry, + TransformStatus, + TrustedTransform, +) +from services.transform_runner import TransformProcessRunner + + +class TestS35TransformIsolation(unittest.TestCase): + + def setUp(self): + self.registry = MagicMock(spec=TransformRegistry) + self.limits = TransformLimits( + timeout_sec=2.0, max_output_bytes=1024, max_transforms_per_request=1 + ) + self.runner = TransformProcessRunner(self.registry, self.limits) + + def test_process_execution_success(self): + """Test successful execution in a subprocess.""" + # Create a dummy transform module on disk + import tempfile + + with tempfile.NamedTemporaryFile(mode="w", suffix=".py", delete=False) as f: + f.write("def transform(data):\n return {'echo': data['input']}") + module_path = f.name + + try: + # Mock registry to return this module + self.registry.get_transform.return_value = TrustedTransform( + id="test_echo", + label="Echo", + module_path=module_path, + sha256="dummy_hash", + ) + self.registry.verify_integrity.return_value = True + + result = self.runner.execute_transform( + "test_echo", {"input": "hello"}, trace_id="test_s35" + ) + + self.assertEqual(result.status, TransformStatus.SUCCESS.value) + self.assertEqual(result.output, {"echo": "hello"}) + # Ensure it ran in a process? Hard to prove from here without spying on subprocess. + # But the runner uses subprocess.run. + finally: + if os.path.exists(module_path): + os.remove(module_path) + + def test_timeout_enforcement(self): + """Test that slow transforms are killed.""" + import tempfile + + with tempfile.NamedTemporaryFile(mode="w", suffix=".py", delete=False) as f: + f.write( + "import time\ndef transform(data):\n time.sleep(5)\n return {}" + ) + module_path = f.name + + try: + self.registry.get_transform.return_value = TrustedTransform( + id="test_slow", + label="Slow", + module_path=module_path, + sha256="dummy_hash", + ) + self.registry.verify_integrity.return_value = True + + result = self.runner.execute_transform( + "test_slow", {}, trace_id="test_timeout" + ) + + self.assertEqual(result.status, TransformStatus.TIMEOUT.value) + self.assertIn("timeout exceeded", result.error) + finally: + if os.path.exists(module_path): + os.remove(module_path) + + def test_capability_denial_network(self): + """Test that network access is denied (by monkeypatch in worker).""" + import tempfile + + with tempfile.NamedTemporaryFile(mode="w", suffix=".py", delete=False) as f: + f.write( + """ +import socket +def transform(data): + try: + s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + s.connect(('example.com', 80)) + return {'status': 'connected'} + except Exception as e: + return {'error': str(e)} +""" + ) + module_path = f.name + + try: + self.registry.get_transform.return_value = TrustedTransform( + id="test_net", label="Net", module_path=module_path, sha256="dummy_hash" + ) + self.registry.verify_integrity.return_value = True + + result = self.runner.execute_transform( + "test_net", {}, trace_id="test_net_deny" + ) + + # The worker monkeypatches socket to raise RuntimeError + # So the transform returns {'error': ...} + # Or crashes if it didn't catch it. + # The script above catches it and returns it. + + output = result.output or {} + self.assertEqual(result.status, TransformStatus.SUCCESS.value) + self.assertIn("Network access denied", output.get("error", "")) + finally: + if os.path.exists(module_path): + os.remove(module_path) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_s36s37r79_egress_hardening.py b/tests/test_s36s37r79_egress_hardening.py index 1d4e6a5..dda2b05 100644 --- a/tests/test_s36s37r79_egress_hardening.py +++ b/tests/test_s36s37r79_egress_hardening.py @@ -19,8 +19,27 @@ from unittest.mock import ANY, MagicMock, patch import services.safe_io as safe_io import services.webhook_auth as webhook_auth -# Disable logging during tests to avoid noise -logging.disable(logging.CRITICAL) +# IMPORTANT: +# Do NOT disable logging at import time. unittest discovery imports all modules +# before test execution, so import-time logging.disable() leaks globally and +# breaks unrelated assertLogs tests. +_PREV_LOG_DISABLE_LEVEL = None + + +def setUpModule(): + """Disable noisy logs only while this module's tests are running.""" + global _PREV_LOG_DISABLE_LEVEL + _PREV_LOG_DISABLE_LEVEL = logging.root.manager.disable + logging.disable(logging.CRITICAL) + + +def tearDownModule(): + """Restore global logging state for downstream modules.""" + global _PREV_LOG_DISABLE_LEVEL + if _PREV_LOG_DISABLE_LEVEL is None: + logging.disable(logging.NOTSET) + else: + logging.disable(_PREV_LOG_DISABLE_LEVEL) class TestS36WebhookReplay(unittest.TestCase):