T05 · Unauthorized Access and Privilege Escalation
Error
- Location
- __init__.py:107
- Finding
- Path Traversal Enables Filesystem Access Outside Relay Storage<![CDATA[ ## Vulnerability Details **File Location**: `__init__.py:74-81`, `__init__.py:107-113`, `__init__.py:247-252`, `__init__.py:309-351`, `__init__.py:590-615`, `run_relay.py:24-34` **Vulnerability Type**: Path traversal and insufficient filesystem boundary validation **Risk Level**: High ### Vulnerable Code ```python def get_event_file(event_id: str) -> Path: """获取事件文件路径""" return STORAGE_PATH / f"{event_id}.json" ``` ```python def resolve_alias(ptr: str, storage_root: Path, alias: str = "s") -> Path: """解析别名路径到完整路径""" if ptr.startswith(f"{alias}/"): return storage_root / ptr[len(alias)+1:] return Path(ptr) ``` ```python file_name = f"{event_id}.json" file_path = STORAGE_PATH / file_name with open(file_path, 'w', encoding='utf-8') as f: json.dump(file_content, f, ensure_ascii=False, indent=2) ``` ```python event_id = parsed["event_id"] ptr = parsed["ptr"] file_path = resolve_alias(ptr, STORAGE_PATH, STORAGE_ALIAS) if not file_path.exists(): raise FileNotFoundError(f"File not found: {file_path}") with open(file_path, 'r', encoding='utf-8') as f: data = json.load(f) content = data.get("payload", {}).get("content", {}) secret = data.get("meta", {}).get("secret", "") meta = data.get("meta", {}) sender = meta.get("sender") receiver = meta.get("receiver") if not sender or not receiver: raise ValueError( f"Event {event_id} is missing explicit sender/receiver metadata" ) if meta.get("burn_on_read") or content.get("burn_on_read"): file_path.unlink(missing_ok=True) ``` ```python target_event_id = next_event_id if next_event_id else event_id file_path = STORAGE_PATH / f"{target_event_id}.json" if not file_path.exists(): raise FileNotFoundError(f"File not found: {file_path}") with open(file_path, 'r', encoding='utf-8') as f: data = json.load(f) if "payload" not in data: data["payload"] = {} if "content" not in data["payload"]: data["payload"]["content"] = {} data["payload"] ...[truncated 3444 chars]
- Remediation
- <![CDATA[ ## Remediation Suggestions 1. Validate every event identifier with a strict allowlist, for example: ```python import re EVENT_ID_PATTERN = re.compile(r"^[A-Za-z0-9_-]{1,128}$") def validate_event_id(event_id: str) -> str: if not EVENT_ID_PATTERN.fullmatch(event_id): raise ValueError("Invalid event ID") return event_id ``` 2. Reject absolute pointers, empty path segments, parent-directory components, and platform-specific path prefixes. 3. Canonicalize and enforce containment before every read, write, or deletion: ```python def safe_storage_path(relative_name: str) -> Path: root = STORAGE_PATH.resolve() candidate = (root / relative_name).resolve() try: candidate.relative_to(root) except ValueError: raise ValueError("Pointer escapes relay storage") return candidate ``` 4. Apply the same validation consistently to send, receive, CMP, verification, update, and cleanup operations. 5. Require the pointer filename to match the validated message event ID rather than trusting them independently. 6. Reject symbolic links or open files using no-follow semantics where supported. Recheck containment after resolving links. 7. Separate read, update, and delete capabilities. Burn-on-read deletion should only be allowed for files that were created and registered by AgentRelay. 8. Add regression tests for absolute paths, `../` traversal, nested traversal, symbolic links, mismatched pointer/event IDs, and platform-specific separators. ]]>