From db7fe477f6badd35cef6feb37dcd32e7bec666e9 Mon Sep 17 00:00:00 2001 From: rookiestar28 <151893693+rookiestar28@users.noreply.github.com> Date: Wed, 18 Feb 2026 01:23:02 +0800 Subject: [PATCH] feat(security): close out Wave B hardening with mTLS binding, pack integrity guards, DoS lifecycle controls, and release provenance verification --- RELEASE_CHECKLIST.md | 10 +++ scripts/generate_provenance.py | 88 +++++++++++++++++++ scripts/verify_provenance.py | 84 ++++++++++++++++++ services/idempotency_store.py | 5 +- services/packs/pack_archive.py | 18 ++-- services/packs/pack_manifest.py | 16 ++++ services/rate_limit.py | 3 + services/sidecar/auth.py | 71 ++++++++++++++- services/tool_runner.py | 17 +++- tests/test_r93_session_invalidation.py | 79 ++++++++++------- tests/test_s50_durable_idempotency.py | 36 ++++---- tests/test_s51_outbound_policy.py | 34 +++---- tests/test_wave_b_r100_provenance.py | 74 ++++++++++++++++ tests/test_wave_b_r101_storage_dos.py | 35 ++++++++ tests/test_wave_b_r104_mtls.py | 81 +++++++++++++++++ .../test_wave_b_s52_manifest_completeness.py | 41 +++++++++ tests/test_wave_b_s53_zipslip.py | 38 ++++++++ tests/test_wave_b_s54.py | 53 +++++++++++ 18 files changed, 705 insertions(+), 78 deletions(-) create mode 100644 scripts/generate_provenance.py create mode 100644 scripts/verify_provenance.py create mode 100644 tests/test_wave_b_r100_provenance.py create mode 100644 tests/test_wave_b_r101_storage_dos.py create mode 100644 tests/test_wave_b_r104_mtls.py create mode 100644 tests/test_wave_b_s52_manifest_completeness.py create mode 100644 tests/test_wave_b_s53_zipslip.py create mode 100644 tests/test_wave_b_s54.py diff --git a/RELEASE_CHECKLIST.md b/RELEASE_CHECKLIST.md index 4fcf9a6..1ce7d93 100644 --- a/RELEASE_CHECKLIST.md +++ b/RELEASE_CHECKLIST.md @@ -65,6 +65,16 @@ npm test --- +## Gate C: Supply Chain Provenance (R100) + +**Goal**: Ensure integrity and traceability of release artifacts. + +- [ ] **Provenance Generation**: Run `python scripts/generate_provenance.py dist/ dist/provenance.json` to create manifest and SBOM. +- [ ] **Verification**: Run `python scripts/verify_provenance.py dist/ dist/provenance.json` on staging environment to verify integrity and completeness. +- [ ] **Completeness**: Ensure `provenance.json` contains SHA256 for all distributed wheels/zips and correct git commit hash. + +--- + ## Release Metadata - [ ] **Version**: `pyproject.toml` version matches git tag. diff --git a/scripts/generate_provenance.py b/scripts/generate_provenance.py new file mode 100644 index 0000000..56357d0 --- /dev/null +++ b/scripts/generate_provenance.py @@ -0,0 +1,88 @@ +#!/usr/bin/env python3 +""" +R100: Generate Release Provenance +Generates a JSON manifest containing: +1. SHA256 checksums of all artifacts in the target directory. +2. Git commit hash (provenance). +3. Build timestamp. +4. SBOM (pip list). + +Usage: + python scripts/generate_provenance.py +""" + +import argparse +import hashlib +import json +import os +import subprocess +import time +from typing import Any, Dict + + +def compute_sha256(file_path: str) -> str: + sha256_hash = hashlib.sha256() + with open(file_path, "rb") as f: + for byte_block in iter(lambda: f.read(4096), b""): + sha256_hash.update(byte_block) + return sha256_hash.hexdigest() + + +def get_git_commit() -> str: + try: + return ( + subprocess.check_output(["git", "rev-parse", "HEAD"]) + .decode("utf-8") + .strip() + ) + except Exception: + return "unknown" + + +def get_sbom() -> Any: + try: + # Simple SBOM: pip list + output = subprocess.check_output(["pip", "list", "--format=json"]).decode( + "utf-8" + ) + return json.loads(output) + except Exception: + return [] + + +def main(): + parser = argparse.ArgumentParser(description="Generate Release Provenance") + parser.add_argument("artifact_dir", help="Directory containing release artifacts") + parser.add_argument("output_json", help="Path to write provenance.json") + args = parser.parse_args() + + artifacts = {} + if os.path.isdir(args.artifact_dir): + for root, _, files in os.walk(args.artifact_dir): + for file in files: + if file == "provenance.json": + continue + full_path = os.path.join(root, file) + rel_path = os.path.relpath(full_path, args.artifact_dir).replace( + "\\", "/" + ) + artifacts[rel_path] = compute_sha256(full_path) + + provenance = { + "meta": { + "timestamp": time.time(), + "git_commit": get_git_commit(), + "generator": "scripts/generate_provenance.py", + }, + "artifacts": artifacts, + "sbom": get_sbom(), + } + + with open(args.output_json, "w", encoding="utf-8") as f: + json.dump(provenance, f, indent=2, sort_keys=True) + + print(f"Provenance generated at {args.output_json}") + + +if __name__ == "__main__": + main() diff --git a/scripts/verify_provenance.py b/scripts/verify_provenance.py new file mode 100644 index 0000000..e7f6112 --- /dev/null +++ b/scripts/verify_provenance.py @@ -0,0 +1,84 @@ +#!/usr/bin/env python3 +""" +R100: Verify Release Provenance +Verifies that artifacts in the directory match the SHA256 checksums in provenance.json. +Fail-safe: Any file existing in dir but missing from provenance is an ERROR. + +Usage: + python scripts/verify_provenance.py +""" + +import argparse +import hashlib +import json +import os +import sys + + +def compute_sha256(file_path: str) -> str: + sha256_hash = hashlib.sha256() + with open(file_path, "rb") as f: + for byte_block in iter(lambda: f.read(4096), b""): + sha256_hash.update(byte_block) + return sha256_hash.hexdigest() + + +def main(): + parser = argparse.ArgumentParser(description="Verify Release Provenance") + parser.add_argument("artifact_dir", help="Directory containing release artifacts") + parser.add_argument("provenance_json", help="Path to provenance.json") + args = parser.parse_args() + + if not os.path.exists(args.provenance_json): + print(f"ERROR: Provenance file not found: {args.provenance_json}") + sys.exit(1) + + with open(args.provenance_json, "r", encoding="utf-8") as f: + provenance = json.load(f) + + expected_artifacts = provenance.get("artifacts", {}) + + # 1. Verify existence and checksums + for rel_path, expected_sha in expected_artifacts.items(): + full_path = os.path.join(args.artifact_dir, rel_path) + if not os.path.exists(full_path): + print(f"ERROR: Missing artifact: {rel_path}") + sys.exit(1) + + actual_sha = compute_sha256(full_path) + if actual_sha != expected_sha: + print(f"ERROR: Checksum mismatch for {rel_path}") + print(f" Expected: {expected_sha}") + print(f" Actual: {actual_sha}") + sys.exit(1) + + # 2. Verify no unlisted files (Completeness) + if os.path.isdir(args.artifact_dir): + for root, _, files in os.walk(args.artifact_dir): + for file in files: + if file == "provenance.json": + continue # provenance might be in the same dir + + full_path = os.path.join(root, file) + rel_path = os.path.relpath(full_path, args.artifact_dir).replace( + "\\", "/" + ) + + # If provenance is inside the dir, we might skip it, but generally provenance is separate or included. + # If checking external provenance file against a dir, provenance file itself isn't checked. + if os.path.abspath(full_path) == os.path.abspath(args.provenance_json): + continue + + if rel_path not in expected_artifacts: + print( + f"ERROR: Unlisted artifact found: {rel_path} (Not in provenance)" + ) + sys.exit(1) + + print("SUCCESS: specific artifacts verified against provenance.") + print(f" Verified {len(expected_artifacts)} artifacts.") + print(f" Git Commit: {provenance.get('meta', {}).get('git_commit')}") + + +if __name__ == "__main__": + main() diff --git a/services/idempotency_store.py b/services/idempotency_store.py index df0a4ec..b4d25aa 100644 --- a/services/idempotency_store.py +++ b/services/idempotency_store.py @@ -247,6 +247,10 @@ class IdempotencyStore: S50: Delegates to durable backend when available. strict_mode: fail-closed if durable backend is unavailable. """ + # R101: Ensure cleanup runs periodically to prevent storage DoS + # This uses an internal timer (300s) to avoid excessive cleanup calls + self._cleanup() + # S50: strict_mode fail-closed if self._strict_mode and not self._durable: raise IdempotencyStoreError( @@ -269,7 +273,6 @@ class IdempotencyStore: ) from e # In-memory path - self._cleanup() now = time.time() with self._store_lock: diff --git a/services/packs/pack_archive.py b/services/packs/pack_archive.py index 95a3e4b..820204b 100644 --- a/services/packs/pack_archive.py +++ b/services/packs/pack_archive.py @@ -2,6 +2,7 @@ import json import os import shutil import tempfile +import unicodedata import zipfile from typing import Optional @@ -151,13 +152,20 @@ class PackArchive: # 2. Safety Check for info in infos: + # S53: Unicode normalization to prevent homoglyph attacks (e.g. fullwidth dots) + # Normalize to NFKC to catch compatibility characters like '．．' -> '..' + norm_name = unicodedata.normalize("NFKC", info.filename) + if ( - info.filename.startswith("/") - or ".." in info.filename - or "\\" in info.filename - or any(c < " " for c in info.filename) # Control chars + norm_name.startswith("/") + or ".." in norm_name + or "\\" in norm_name + or ":" in norm_name # Block drive-relative paths (C:foo) + or any(c < " " for c in norm_name) # Control chars ): - raise PackError(f"Unsafe filename: {info.filename}") + raise PackError( + f"Unsafe filename: {info.filename} (normalized: {norm_name})" + ) # Check for symlinks (S_IFLNK - 0xA000) # ZipInfo.external_attr: upper 16 bits are Unix permissions diff --git a/services/packs/pack_manifest.py b/services/packs/pack_manifest.py index 3d828df..70788ff 100644 --- a/services/packs/pack_manifest.py +++ b/services/packs/pack_manifest.py @@ -60,8 +60,10 @@ def validate_manifest_integrity(base_dir: str, manifest: PackManifest) -> List[s if len(manifest.get("files", [])) > MAX_MANIFEST_FILES: return [f"Manifest exceeds maximum file count ({MAX_MANIFEST_FILES})"] + manifest_paths = set() for item in manifest.get("files", []): rel_path = item.get("path") + manifest_paths.add(rel_path) expected_hash = item.get("sha256") # S4: Path Traversal Check (Redundant but critical) @@ -89,6 +91,20 @@ def validate_manifest_integrity(base_dir: str, manifest: PackManifest) -> List[s except Exception as e: errors.append(f"Error reading {rel_path}: {str(e)}") + # S52: Manifest Completeness Check + # Ensure no files on disk are missing from manifest (except metadata files) + allowed_extras = {"manifest.json", "pack.json"} + for root, _, files in os.walk(base_dir): + for file in files: + full_path = os.path.join(root, file) + rel_path = os.path.relpath(full_path, base_dir).replace("\\", "/") + + if rel_path in allowed_extras: + continue + + if rel_path not in manifest_paths: + errors.append(f"Unlisted file found: {rel_path} (S52 Violations)") + return errors diff --git a/services/rate_limit.py b/services/rate_limit.py index 29f42e9..d09d3f0 100644 --- a/services/rate_limit.py +++ b/services/rate_limit.py @@ -58,6 +58,9 @@ class RateLimiter: "logs": (60, 60.0 / 60.0), # 60 req/min "admin": (20, 20.0 / 60.0), # 20 req/min "bridge": (20, 20.0 / 60.0), # 20 req/min + # R101: New Quotas + "connector": (20, 20.0 / 60.0), # 20 req/min (aligned with bridge) + "trigger": (60, 60.0 / 60.0), # 60 req/min (higher due to automation) } def check(self, limit_type: str, ip: str) -> bool: diff --git a/services/sidecar/auth.py b/services/sidecar/auth.py index 81ab23d..2e7b3fe 100644 --- a/services/sidecar/auth.py +++ b/services/sidecar/auth.py @@ -8,7 +8,7 @@ from __future__ import annotations import hmac import logging import os -from typing import Optional, Set, Tuple +from typing import Dict, Optional, Set, Tuple try: from aiohttp import web # type: ignore @@ -26,6 +26,11 @@ ENV_BRIDGE_DEVICE_TOKEN = "OPENCLAW_BRIDGE_DEVICE_TOKEN" LEGACY_ENV_BRIDGE_DEVICE_TOKEN = "MOLTBOT_BRIDGE_DEVICE_TOKEN" ENV_BRIDGE_ALLOWED_DEVICE_IDS = "OPENCLAW_BRIDGE_ALLOWED_DEVICE_IDS" LEGACY_ENV_BRIDGE_ALLOWED_DEVICE_IDS = "MOLTBOT_BRIDGE_ALLOWED_DEVICE_IDS" +# R104: mTLS Contract +ENV_BRIDGE_MTLS_ENABLED = "OPENCLAW_BRIDGE_MTLS_ENABLED" +ENV_BRIDGE_DEVICE_CERT_MAP = ( + "OPENCLAW_BRIDGE_DEVICE_CERT_MAP" # device_id:fingerprint,... +) # Headers HEADER_DEVICE_ID = "X-OpenClaw-Device-Id" @@ -34,6 +39,8 @@ HEADER_DEVICE_TOKEN = "X-OpenClaw-Device-Token" LEGACY_HEADER_DEVICE_TOKEN = "X-Moltbot-Device-Token" HEADER_SCOPES = "X-OpenClaw-Scopes" LEGACY_HEADER_SCOPES = "X-Moltbot-Scopes" +# R104: mTLS Headers +HEADER_CLIENT_CERT_HASH = "X-Client-Cert-Hash" # SHA256 fingerprint from proxy def _env_get(primary: str, legacy: str, default: str = "") -> str: @@ -72,6 +79,62 @@ def get_allowed_device_ids() -> Optional[Set[str]]: return set(id.strip() for id in ids_str.split(",") if id.strip()) +def is_mtls_enabled() -> bool: + """Check if mTLS enforcement is enabled.""" + return os.environ.get(ENV_BRIDGE_MTLS_ENABLED, "false").lower() in ( + "true", + "1", + "yes", + "on", + ) + + +def get_device_cert_map() -> Dict[str, str]: + """ + Get map of device_id -> certificate fingerprint. + Format: device_id:fingerprint,device_id2:fingerprint2 + """ + mapping_str = os.environ.get(ENV_BRIDGE_DEVICE_CERT_MAP, "") + if not mapping_str: + return {} + + result = {} + for entry in mapping_str.split(","): + if ":" in entry: + parts = entry.split(":", 1) + result[parts[0].strip()] = parts[1].strip() + return result + + +def validate_mtls_binding(request: web.Request, device_id: str) -> Tuple[bool, str]: + """ + R104: Validate mTLS certificate binding for the device. + """ + if not is_mtls_enabled(): + return True, "" + + cert_hash = request.headers.get(HEADER_CLIENT_CERT_HASH, "") + if not cert_hash: + # Strict mode: mTLS enabled but no cert header -> fail + return False, "Missing client certificate header (mTLS required)" + + cert_map = get_device_cert_map() + expected_hash = cert_map.get(device_id) + + if not expected_hash: + # Strict mode: mTLS enabled implies explicit device binding + return False, f"Device not bound to a certificate (Device ID: {device_id})" + + # Constant-time comparison not strictly required for public fingerprints but good practice + if not hmac.compare_digest(cert_hash, expected_hash): + logger.warning( + f"mTLS violation: Device {device_id} presented {cert_hash}, expected {expected_hash}" + ) + return False, "Certificate fingerprint mismatch" + + return True, "" + + def validate_device_token( request: web.Request, required_scope: Optional[BridgeScope] = None ) -> Tuple[bool, str, Optional[str]]: @@ -138,6 +201,12 @@ def validate_device_token( ) return False, f"Missing required scope: {required_scope}", None + # R104: mTLS Binding Check + is_mtls_valid, mtls_error = validate_mtls_binding(request, device_id) + if not is_mtls_valid: + logger.warning(f"mTLS validation failed for {device_id}: {mtls_error}") + return False, mtls_error, None + return True, "", device_id diff --git a/services/tool_runner.py b/services/tool_runner.py index 7ac6ab8..2ece4c1 100644 --- a/services/tool_runner.py +++ b/services/tool_runner.py @@ -76,10 +76,19 @@ class SandboxProfile: return [] violations: List[str] = [] for p in paths: - resolved = os.path.abspath(p) - if not any( - resolved.startswith(os.path.abspath(allowed)) for allowed in allowlist - ): + # S54: Canonicalize path to resolve symlinks and '..' + resolved = os.path.realpath(p) + + # Check against allowed prefixes + is_allowed = False + for allowed in allowlist: + allowed_abs = os.path.realpath(allowed) + # Ensure we match directory boundary (exact match or subdir) + if resolved == allowed_abs or resolved.startswith(allowed_abs + os.sep): + is_allowed = True + break + + if not is_allowed: mode = "write" if write else "read" violations.append( f"Path '{resolved}' not in allow_fs_{mode}: {allowlist}" diff --git a/tests/test_r93_session_invalidation.py b/tests/test_r93_session_invalidation.py index ee08029..9339d65 100644 --- a/tests/test_r93_session_invalidation.py +++ b/tests/test_r93_session_invalidation.py @@ -1,46 +1,58 @@ - import unittest from unittest.mock import AsyncMock, MagicMock, patch -from connector.transport_contract import RelayResponseClassifier, RelayStatus -from connector.platforms.whatsapp_webhook import WhatsAppWebhookServer from connector.platforms.line_webhook import LINEWebhookServer from connector.platforms.wechat_webhook import WeChatWebhookServer +from connector.platforms.whatsapp_webhook import WhatsAppWebhookServer from connector.router import CommandRouter +from connector.transport_contract import RelayResponseClassifier, RelayStatus + class TestRelayResponseClassifier(unittest.TestCase): def test_auth_invalid_codes(self): """Test that 401 and 410 are classified as auth_invalid.""" print(f"DEBUG: classify(401) = {RelayResponseClassifier.classify(401)}") - print(f"DEBUG: AUTH_INVALID_CODES = {RelayResponseClassifier.AUTH_INVALID_CODES}") + print( + f"DEBUG: AUTH_INVALID_CODES = {RelayResponseClassifier.AUTH_INVALID_CODES}" + ) self.assertTrue(RelayResponseClassifier.is_auth_invalid(401)) self.assertTrue(RelayResponseClassifier.is_auth_invalid(410)) - self.assertEqual(RelayResponseClassifier.classify(401), RelayStatus.AUTH_INVALID) - self.assertEqual(RelayResponseClassifier.classify(410), RelayStatus.AUTH_INVALID) + self.assertEqual( + RelayResponseClassifier.classify(401), RelayStatus.AUTH_INVALID + ) + self.assertEqual( + RelayResponseClassifier.classify(410), RelayStatus.AUTH_INVALID + ) def test_other_codes(self): """Test that other codes are not auth_invalid.""" self.assertFalse(RelayResponseClassifier.is_auth_invalid(200)) - self.assertFalse(RelayResponseClassifier.is_auth_invalid(403)) # Forbidden != Auth Invalid (usually) + self.assertFalse( + RelayResponseClassifier.is_auth_invalid(403) + ) # Forbidden != Auth Invalid (usually) self.assertFalse(RelayResponseClassifier.is_auth_invalid(500)) # Use .value to avoid enum identity issues if any - self.assertEqual(RelayResponseClassifier.classify(200).value, RelayStatus.OK.value) + self.assertEqual( + RelayResponseClassifier.classify(200).value, RelayStatus.OK.value + ) # 500 is TRANSIENT in OpenClaw (retryable), not SERVER_ERROR (fatal) - self.assertEqual(RelayResponseClassifier.classify(500).value, RelayStatus.TRANSIENT.value) + self.assertEqual( + RelayResponseClassifier.classify(500).value, RelayStatus.TRANSIENT.value + ) + class TestWebhookSessionInvalidation(unittest.IsolatedAsyncioTestCase): async def test_whatsapp_session_invalidation(self): """Test WhatsApp session invalidation on 401.""" server = WhatsAppWebhookServer( - config=MagicMock(), - router=MagicMock(spec=CommandRouter) + config=MagicMock(), router=MagicMock(spec=CommandRouter) ) server.session = MagicMock() - + # Mock response context manager mock_response = AsyncMock() mock_response.status = 401 - + # Mock post context manager post_ctx = AsyncMock() post_ctx.__aenter__.return_value = mock_response @@ -49,10 +61,10 @@ class TestWebhookSessionInvalidation(unittest.IsolatedAsyncioTestCase): # First call should trigger invalidation await server.send_message("123", "test") self.assertTrue(server._session_invalid) - + # Reset mock to verify no further calls server.session.post.reset_mock() - + # Second call should be blocked await server.send_message("123", "test") server.session.post.assert_not_called() @@ -60,24 +72,23 @@ class TestWebhookSessionInvalidation(unittest.IsolatedAsyncioTestCase): async def test_line_session_invalidation(self): """Test LINE session invalidation on 401.""" server = LINEWebhookServer( - config=MagicMock(), - router=MagicMock(spec=CommandRouter) + config=MagicMock(), router=MagicMock(spec=CommandRouter) ) server.session = MagicMock() - + mock_response = AsyncMock() mock_response.status = 401 - + post_ctx = AsyncMock() post_ctx.__aenter__.return_value = mock_response server.session.post.return_value = post_ctx - + # Mock _get_header to avoid errors server._get_header = MagicMock(return_value={}) await server._reply_message("token", "test") self.assertTrue(server._session_invalid) - + server.session.post.reset_mock() await server._reply_message("token", "test") server.session.post.assert_not_called() @@ -85,15 +96,14 @@ class TestWebhookSessionInvalidation(unittest.IsolatedAsyncioTestCase): async def test_wechat_session_invalidation(self): """Test WeChat session invalidation on 401 (token fetch).""" server = WeChatWebhookServer( - config=MagicMock(), - router=MagicMock(spec=CommandRouter) + config=MagicMock(), router=MagicMock(spec=CommandRouter) ) server.session = MagicMock() - + # Mock token fetch response mock_response = AsyncMock() mock_response.status = 401 - + get_ctx = AsyncMock() get_ctx.__aenter__.return_value = mock_response server.session.get.return_value = get_ctx @@ -101,39 +111,40 @@ class TestWebhookSessionInvalidation(unittest.IsolatedAsyncioTestCase): token = await server._get_access_token() self.assertIsNone(token) self.assertTrue(server._session_invalid) - + # Reset mock and try again server.session.get.reset_mock() - + # Second call should be blocked by _session_invalid check (which must be inside access_token logic) # Wait - _get_access_token doesn't check _session_invalid at the start? # Let's check implementation. If not, we should better add it or test assumes it. # But wait, self._session_invalid = True was set. # If the implementation doesn't check it at start of _get_access_token, it will try to fetch again? # The test expects assert_not_called. - + # Let's enforce it in the test by ensuring the method respects the flag OR # duplicate the session invalidation check in _get_access_token (which is smart). - + # If we didn't add the check in _get_access_token, let's just assert that it Returns None immediately? # Actually proper R93 implies *locking* the connector. - + # Let's see... if I look at the previous step, I only added the classification logic. # I did not add "if self._session_invalid: return None" at the top of _get_access_token. # So it probably TRIED to fetch again. - + # We should probably add that check to _get_access_token too for completeness of R93. - # But for now let's fix the test expectation if we want to rely on the *result* being None + # But for now let's fix the test expectation if we want to rely on the *result* being None # (and maybe it fetches again? No, we want to STOP traffic). - + # Let's update the test to be realistic or update the code. # R93 says "Connector platforms should stop retrying". # So _get_access_token SHOULD check the flag. - + # I will update the code in next step. For now let's update test to expect it to be blocked. token2 = await server._get_access_token() server.session.get.assert_not_called() self.assertIsNone(token2) + if __name__ == "__main__": unittest.main() diff --git a/tests/test_s50_durable_idempotency.py b/tests/test_s50_durable_idempotency.py index 875e0e3..752ccd4 100644 --- a/tests/test_s50_durable_idempotency.py +++ b/tests/test_s50_durable_idempotency.py @@ -1,19 +1,19 @@ - import os +import shutil import sqlite3 +import tempfile import time import unittest -import tempfile -import shutil from unittest.mock import MagicMock, patch from services.idempotency_store import ( - IdempotencyStore, - SQLiteDurableBackend, DurableBackend, - IdempotencyStoreError + IdempotencyStore, + IdempotencyStoreError, + SQLiteDurableBackend, ) + class TestSQLiteDurableBackend(unittest.TestCase): def setUp(self): self.temp_dir = tempfile.mkdtemp() @@ -25,7 +25,7 @@ class TestSQLiteDurableBackend(unittest.TestCase): self.backend.close() except: pass - + # Retry cleanup for Windows file locking for _ in range(5): try: @@ -45,18 +45,18 @@ class TestSQLiteDurableBackend(unittest.TestCase): backend2 = SQLiteDurableBackend(self.db_path) # If key exists, it returns True (fresh) if expired, or False (dup) if valid? # check_and_record returns (True, pid) if existing - # WAIT: Protocol says: + # WAIT: Protocol says: # Check if key exists; if not, record it. Returns (not_exists, existing_prompt_id) ?? # Docstring: "Returns (is_dup, existing_prompt_id)." in Protocol. # Impl: if row: return True, existing_pid (meaning IS DUP). # Impl: if not row: insert, return False, None (meaning NOT DUP). - + # So duplicate -> True. is_dup, pid = backend2.check_and_record("key1", 3600) self.assertTrue(is_dup) self.assertEqual(pid, "prompt1") finally: - if 'backend2' in locals(): + if "backend2" in locals(): backend2.close() def test_ttl_expiry(self): @@ -64,15 +64,16 @@ class TestSQLiteDurableBackend(unittest.TestCase): # Insert with short TTL self.backend.check_and_record("key_ttl", 1) time.sleep(2) # Wait for expiry - + # Cleanup should remove it self.backend.cleanup() - + # Should be fresh again # Impl: if fresh, returns (False, None) -> is_dup=False is_dup, pid = self.backend.check_and_record("key_ttl", 3600) self.assertFalse(is_dup) + class TestIdempotencyStoreS50(unittest.TestCase): def setUp(self): self.temp_dir = tempfile.mkdtemp() @@ -88,11 +89,11 @@ class TestIdempotencyStoreS50(unittest.TestCase): """Test fail-closed behavior in strict mode.""" mock_backend = MagicMock(spec=DurableBackend) mock_backend.check_and_record.side_effect = Exception("Disk failure") - + # Singleton instantiation store = IdempotencyStore() store.configure_durable(backend=mock_backend, strict_mode=True) - + with self.assertRaises(IdempotencyStoreError): store.check_and_record("test_key", ttl=60) @@ -100,17 +101,17 @@ class TestIdempotencyStoreS50(unittest.TestCase): """Test fallback behavior in lenient mode.""" mock_backend = MagicMock(spec=DurableBackend) mock_backend.check_and_record.side_effect = Exception("Disk failure") - + store = IdempotencyStore() store.configure_durable(backend=mock_backend, strict_mode=False) - + # Should NOT raise, but log error and fallback (or just return False? or True?) # Implementation of IdempotencyStore in strict_mode=False absorbs errors? # Let's check implementation behavior assumption: # If backend fails, and not strict, it might default to "allow" (True) or "deny"? # Actually existing implementation likely just logs and returns True (allow execution) or False? # Typically fail-open for availability means returning True (fresh). - + # Taking a peek at IdempotencyStore implementation would help, but assuming fail-open for now based on standard patterns. # If it raises, I'll fix the test. try: @@ -120,5 +121,6 @@ class TestIdempotencyStoreS50(unittest.TestCase): except IdempotencyStoreError: self.fail("Should not raise IdempotencyStoreError in lenient mode") + if __name__ == "__main__": unittest.main() diff --git a/tests/test_s51_outbound_policy.py b/tests/test_s51_outbound_policy.py index 1102720..e5244cf 100644 --- a/tests/test_s51_outbound_policy.py +++ b/tests/test_s51_outbound_policy.py @@ -1,44 +1,45 @@ - import unittest + from services.safe_io import ( - OutboundPolicy, - validate_outbound_url, - SSRFError, + STANDARD_OUTBOUND_POLICY, STRICT_OUTBOUND_POLICY, - STANDARD_OUTBOUND_POLICY + OutboundPolicy, + SSRFError, + validate_outbound_url, ) + class TestOutboundPolicyS51(unittest.TestCase): def test_strict_policy(self): """Test STRICT_OUTBOUND_POLICY enforcement.""" policy = STRICT_OUTBOUND_POLICY - + # HTTPS 443 -> OK self.assertIsNone(policy.validate("https", 443)) - + # HTTP -> Fail self.assertIsNotNone(policy.validate("http", 80)) - + # Custom Port -> Fail self.assertIsNotNone(policy.validate("https", 8443)) def test_standard_policy(self): """Test STANDARD_OUTBOUND_POLICY enforcement.""" policy = STANDARD_OUTBOUND_POLICY - + # HTTP 80 -> OK self.assertIsNone(policy.validate("http", 80)) - + # HTTPS 443 -> OK self.assertIsNone(policy.validate("https", 443)) - + # 8080/8443 -> OK self.assertIsNone(policy.validate("http", 8080)) self.assertIsNone(policy.validate("https", 8443)) - + # Ollama 11434 -> OK (as updated) self.assertIsNone(policy.validate("http", 11434)) - + # Random port -> Fail self.assertIsNotNone(policy.validate("http", 9999)) @@ -48,7 +49,7 @@ class TestOutboundPolicyS51(unittest.TestCase): scheme, host, port, ips = validate_outbound_url( "https://google.com", allow_any_public_host=True, - policy=STRICT_OUTBOUND_POLICY + policy=STRICT_OUTBOUND_POLICY, ) self.assertEqual(scheme, "https") self.assertEqual(port, 443) @@ -58,7 +59,7 @@ class TestOutboundPolicyS51(unittest.TestCase): validate_outbound_url( "http://google.com", allow_any_public_host=True, - policy=STRICT_OUTBOUND_POLICY + policy=STRICT_OUTBOUND_POLICY, ) self.assertIn("S51", str(cm.exception)) @@ -67,9 +68,10 @@ class TestOutboundPolicyS51(unittest.TestCase): validate_outbound_url( "http://google.com:9999", allow_any_public_host=True, - policy=STANDARD_OUTBOUND_POLICY + policy=STANDARD_OUTBOUND_POLICY, ) self.assertIn("S51", str(cm.exception)) + if __name__ == "__main__": unittest.main() diff --git a/tests/test_wave_b_r100_provenance.py b/tests/test_wave_b_r100_provenance.py new file mode 100644 index 0000000..56136a9 --- /dev/null +++ b/tests/test_wave_b_r100_provenance.py @@ -0,0 +1,74 @@ +import json +import os +import shutil +import subprocess +import sys +import tempfile +import unittest +from pathlib import Path + + +class TestR100Provenance(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.mkdtemp(prefix="wave_b_r100_") + self.artifacts = os.path.join(self.tmp, "dist") + os.makedirs(self.artifacts, exist_ok=True) + self.provenance_json = os.path.join(self.artifacts, "provenance.json") + with open( + os.path.join(self.artifacts, "artifact_a.txt"), "w", encoding="utf-8" + ) as f: + f.write("artifact-a") + + repo_root = Path(__file__).resolve().parents[1] + self.generate_script = str(repo_root / "scripts" / "generate_provenance.py") + self.verify_script = str(repo_root / "scripts" / "verify_provenance.py") + + def tearDown(self): + shutil.rmtree(self.tmp, ignore_errors=True) + + def _run(self, *args): + return subprocess.run( + [sys.executable, *args], + text=True, + capture_output=True, + check=False, + ) + + def test_provenance_lifecycle(self): + generate = self._run(self.generate_script, self.artifacts, self.provenance_json) + self.assertEqual(generate.returncode, 0, generate.stderr + generate.stdout) + self.assertTrue(os.path.exists(self.provenance_json)) + + with open(self.provenance_json, "r", encoding="utf-8") as f: + payload = json.load(f) + self.assertIn("artifact_a.txt", payload.get("artifacts", {})) + + verify_ok = self._run(self.verify_script, self.artifacts, self.provenance_json) + self.assertEqual(verify_ok.returncode, 0, verify_ok.stderr + verify_ok.stdout) + + with open( + os.path.join(self.artifacts, "artifact_a.txt"), "w", encoding="utf-8" + ) as f: + f.write("tampered") + verify_tampered = self._run( + self.verify_script, self.artifacts, self.provenance_json + ) + self.assertNotEqual(verify_tampered.returncode, 0) + self.assertIn("Checksum mismatch", verify_tampered.stdout) + + def test_unlisted_file_detection(self): + generate = self._run(self.generate_script, self.artifacts, self.provenance_json) + self.assertEqual(generate.returncode, 0, generate.stderr + generate.stdout) + + with open( + os.path.join(self.artifacts, "artifact_b.txt"), "w", encoding="utf-8" + ) as f: + f.write("not listed") + + verify = self._run(self.verify_script, self.artifacts, self.provenance_json) + self.assertNotEqual(verify.returncode, 0) + self.assertIn("Unlisted artifact found", verify.stdout) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_wave_b_r101_storage_dos.py b/tests/test_wave_b_r101_storage_dos.py new file mode 100644 index 0000000..7407911 --- /dev/null +++ b/tests/test_wave_b_r101_storage_dos.py @@ -0,0 +1,35 @@ +import unittest +from unittest.mock import MagicMock + +from services.idempotency_store import IdempotencyStore +from services.rate_limit import RateLimiter + + +class TestR101StorageDoS(unittest.TestCase): + def setUp(self): + IdempotencyStore.reset_singleton() + + def tearDown(self): + IdempotencyStore.reset_singleton() + + def test_cleanup_called_in_durable_mode_eventually(self): + backend = MagicMock() + backend.check_and_record.return_value = (False, None) + + store = IdempotencyStore() + store.configure_durable(backend=backend, strict_mode=False) + store.check_and_record("wave_b_key", ttl=60) + + backend.cleanup.assert_called() + backend.check_and_record.assert_called_once() + + def test_connector_and_trigger_quotas_exist(self): + limiter = RateLimiter() + self.assertIn("connector", limiter.defaults) + self.assertIn("trigger", limiter.defaults) + self.assertEqual(limiter.defaults["connector"][0], 20) + self.assertEqual(limiter.defaults["trigger"][0], 60) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_wave_b_r104_mtls.py b/tests/test_wave_b_r104_mtls.py new file mode 100644 index 0000000..30d1e9a --- /dev/null +++ b/tests/test_wave_b_r104_mtls.py @@ -0,0 +1,81 @@ +import os +import unittest +from unittest.mock import MagicMock + +import services.sidecar.auth as auth_module + + +class TestR104MTLS(unittest.TestCase): + def setUp(self): + self.env_keys = [ + "OPENCLAW_BRIDGE_ENABLED", + "OPENCLAW_BRIDGE_DEVICE_TOKEN", + "OPENCLAW_BRIDGE_ALLOWED_DEVICE_IDS", + "OPENCLAW_BRIDGE_MTLS_ENABLED", + "OPENCLAW_BRIDGE_DEVICE_CERT_MAP", + "MOLTBOT_BRIDGE_ENABLED", + "MOLTBOT_BRIDGE_DEVICE_TOKEN", + "MOLTBOT_BRIDGE_ALLOWED_DEVICE_IDS", + ] + for key in self.env_keys: + os.environ.pop(key, None) + + def tearDown(self): + for key in self.env_keys: + os.environ.pop(key, None) + + def _base_request(self): + req = MagicMock() + req.headers = { + "X-OpenClaw-Device-Id": "dev1", + "X-OpenClaw-Device-Token": "token123", + } + return req + + def _enable_base_auth(self): + os.environ["OPENCLAW_BRIDGE_ENABLED"] = "1" + os.environ["OPENCLAW_BRIDGE_DEVICE_TOKEN"] = "token123" + + def test_mtls_disabled_by_default(self): + self._enable_base_auth() + req = self._base_request() + valid, err, device = auth_module.validate_device_token(req) + self.assertTrue(valid) + self.assertEqual(err, "") + self.assertEqual(device, "dev1") + + def test_mtls_enforced_when_enabled(self): + self._enable_base_auth() + os.environ["OPENCLAW_BRIDGE_MTLS_ENABLED"] = "1" + req = self._base_request() + valid, err, _ = auth_module.validate_device_token(req) + self.assertFalse(valid) + self.assertIn("mTLS required", err) + + def test_mtls_fingerprint_mismatch(self): + self._enable_base_auth() + os.environ["OPENCLAW_BRIDGE_MTLS_ENABLED"] = "1" + os.environ["OPENCLAW_BRIDGE_DEVICE_CERT_MAP"] = "dev1:sha256_expected" + + req = self._base_request() + req.headers["X-Client-Cert-Hash"] = "sha256_actual" + + valid, err, _ = auth_module.validate_device_token(req) + self.assertFalse(valid) + self.assertIn("fingerprint mismatch", err.lower()) + + def test_mtls_strict_unbound_device(self): + self._enable_base_auth() + os.environ["OPENCLAW_BRIDGE_MTLS_ENABLED"] = "1" + os.environ["OPENCLAW_BRIDGE_DEVICE_CERT_MAP"] = "dev2:sha256_for_other_device" + + req = self._base_request() + req.headers["X-Client-Cert-Hash"] = "sha256_actual" + + valid, err, _ = auth_module.validate_device_token(req) + self.assertFalse(valid) + self.assertIn("not bound", err.lower()) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_wave_b_s52_manifest_completeness.py b/tests/test_wave_b_s52_manifest_completeness.py new file mode 100644 index 0000000..2f644ea --- /dev/null +++ b/tests/test_wave_b_s52_manifest_completeness.py @@ -0,0 +1,41 @@ +import hashlib +import os +import shutil +import tempfile +import unittest + +from services.packs.pack_manifest import validate_manifest_integrity + + +class TestS52ManifestCompleteness(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.mkdtemp(prefix="wave_b_s52_") + + def tearDown(self): + shutil.rmtree(self.tmp, ignore_errors=True) + + def test_unlisted_files_rejected(self): + pack_json_path = os.path.join(self.tmp, "pack.json") + with open(pack_json_path, "w", encoding="utf-8") as f: + f.write('{"name":"demo"}') + + with open(pack_json_path, "rb") as f: + pack_hash = hashlib.sha256(f.read()).hexdigest() + + with open(os.path.join(self.tmp, "evil.txt"), "w", encoding="utf-8") as f: + f.write("hidden payload") + + manifest = { + "files": [ + { + "path": "pack.json", + "sha256": pack_hash, + } + ] + } + errors = validate_manifest_integrity(self.tmp, manifest) + self.assertTrue(any("Unlisted file found: evil.txt" in e for e in errors)) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_wave_b_s53_zipslip.py b/tests/test_wave_b_s53_zipslip.py new file mode 100644 index 0000000..cb8d017 --- /dev/null +++ b/tests/test_wave_b_s53_zipslip.py @@ -0,0 +1,38 @@ +import os +import shutil +import tempfile +import unittest +import zipfile + +from services.packs.pack_archive import PackArchive, PackError + + +class TestS53ZipSlip(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.mkdtemp(prefix="wave_b_s53_") + + def tearDown(self): + shutil.rmtree(self.tmp, ignore_errors=True) + + def test_unicode_traversal(self): + zip_path = os.path.join(self.tmp, "unicode_traversal.zip") + # U+FF0E FULLWIDTH FULL STOP x2 normalizes to ".." under NFKC. + unsafe_name = "\uFF0E\uFF0E/evil.txt" + + with zipfile.ZipFile(zip_path, "w") as zf: + zf.writestr(unsafe_name, "payload") + + with self.assertRaisesRegex(PackError, "Unsafe filename"): + PackArchive.extract_pack(zip_path, os.path.join(self.tmp, "out")) + + def test_drive_relative_path_rejected(self): + zip_path = os.path.join(self.tmp, "drive_relative.zip") + with zipfile.ZipFile(zip_path, "w") as zf: + zf.writestr("C:evil.txt", "payload") + + with self.assertRaisesRegex(PackError, "Unsafe filename"): + PackArchive.extract_pack(zip_path, os.path.join(self.tmp, "out2")) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_wave_b_s54.py b/tests/test_wave_b_s54.py new file mode 100644 index 0000000..485ad40 --- /dev/null +++ b/tests/test_wave_b_s54.py @@ -0,0 +1,53 @@ +import os +import shutil +import tempfile +import unittest + +from services.tool_runner import SandboxProfile + + +class TestS54PathTraversal(unittest.TestCase): + def setUp(self): + self.root = tempfile.mkdtemp(prefix="wave_b_s54_") + self.allowed = os.path.join(self.root, "allowed") + self.sibling = os.path.join(self.root, "allowed_evil") + os.makedirs(self.allowed, exist_ok=True) + os.makedirs(self.sibling, exist_ok=True) + + self.profile = SandboxProfile( + network=False, + allow_fs_read=[self.allowed], + allow_fs_write=[self.allowed], + allow_network_hosts=[], + ) + + def tearDown(self): + shutil.rmtree(self.root, ignore_errors=True) + + def test_prefix_bypass(self): + nested_ok = os.path.join(self.allowed, "safe.txt") + sibling_bad = os.path.join(self.sibling, "escape.txt") + + ok = self.profile.validate_fs_access([nested_ok], write=False) + bad = self.profile.validate_fs_access([sibling_bad], write=False) + + self.assertEqual(ok, []) + self.assertTrue(any("not in allow_fs_read" in item for item in bad)) + + def test_symlink_escape(self): + outside_file = os.path.join(self.root, "outside.txt") + with open(outside_file, "w", encoding="utf-8") as f: + f.write("outside") + + link_path = os.path.join(self.allowed, "link_to_outside.txt") + try: + os.symlink(outside_file, link_path) + except (AttributeError, NotImplementedError, OSError): + self.skipTest("Symlink creation not available in this environment") + + violations = self.profile.validate_fs_access([link_path], write=False) + self.assertTrue(any("not in allow_fs_read" in item for item in violations)) + + +if __name__ == "__main__": + unittest.main()