diff --git a/demos/claude_tool_authority/authority.py b/demos/claude_tool_authority/authority.py index ff231841..c3e207fe 100644 --- a/demos/claude_tool_authority/authority.py +++ b/demos/claude_tool_authority/authority.py @@ -7,11 +7,12 @@ the returned ``licensed_action`` is a data artifact, not an execution path. from __future__ import annotations +import copy import hashlib import json import re from functools import lru_cache -from pathlib import Path +from pathlib import Path, PurePosixPath from typing import Any, Final TOOL_NAME: Final[str] = "core.tool_authority.review" @@ -38,6 +39,7 @@ _ENVELOPE_AUTHORITY: Final[str] = "demo_tool_authority_envelope(local-v1)" _LICENSE_AUTHORITY: Final[str] = "demos.claude_tool_authority.authority.build_license" _ALLOWED_NOTE_PREFIX: Final[str] = "workspace/demo_notes/" +_ALLOWED_NOTE_ROOT: Final[PurePosixPath] = PurePosixPath("workspace/demo_notes") _PROTECTED_PATH_PREFIXES: Final[tuple[str, ...]] = ( ".git/", "chat/", @@ -229,6 +231,60 @@ def _protected_target(path: str | None) -> bool: return any(path.startswith(prefix) for prefix in _PROTECTED_PATH_PREFIXES) +def _normalize_note_target_path(raw_path: Any) -> tuple[str | None, str | None]: + """Canonicalize demo note paths under deterministic POSIX semantics. + + The input is JSON/demo artifact data, not a host filesystem path. Be + conservative: any absolute path, empty path, or ``..`` segment is refused. + """ + if not isinstance(raw_path, str): + return None, "malformed_target_path" + if raw_path == "": + return None, "empty_target_path" + + normalized = PurePosixPath(raw_path) + if normalized.is_absolute(): + return None, "absolute_target_path" + + parts = normalized.parts + if not parts: + return None, "malformed_target_path" + if any(part == ".." for part in parts): + return None, "path_traversal_detected" + + cleaned_parts = tuple(part for part in parts if part not in ("", ".")) + if not cleaned_parts: + return None, "malformed_target_path" + + cleaned = PurePosixPath(*cleaned_parts) + cleaned_text = cleaned.as_posix() + if _protected_target(cleaned_text): + return None, "protected_path" + if not cleaned_text.startswith(_ALLOWED_NOTE_PREFIX): + return None, "outside_authority_envelope" + if cleaned == _ALLOWED_NOTE_ROOT: + return None, "malformed_target_path" + if not cleaned.is_relative_to(_ALLOWED_NOTE_ROOT): + return None, "outside_authority_envelope" + return cleaned_text, None + + +def _normalized_payload(payload: dict[str, Any]) -> tuple[dict[str, Any], str | None]: + """Copy and canonicalize caller payload before authority evaluation.""" + normalized = copy.deepcopy(payload) + action_request = normalized["action_request"] + if action_request["action_type"] != "write_local_note": + return normalized, None + + target = action_request["target"] + normalized_path, refusal_reason = _normalize_note_target_path(target.get("path")) + if refusal_reason is not None: + return normalized, refusal_reason + assert normalized_path is not None + target["path"] = normalized_path + return normalized, None + + def build_license(payload: dict[str, Any]) -> dict[str, Any]: target_path = payload["action_request"]["target"]["path"] content = payload["action_request"]["payload"]["content"] @@ -322,7 +378,29 @@ def run_authority(payload: Any) -> dict[str, Any]: if errors: return _invalid_response(payload, errors) assert isinstance(payload, dict) - return evaluate_authority(payload) + normalized_payload, refusal_reason = _normalized_payload(payload) + if refusal_reason is not None: + return _finalize( + { + "tool": TOOL_NAME, + "status": "refused", + "request_id": normalized_payload["request_id"], + "scenario_id": normalized_payload["scenario_id"], + "authority_path": [_ROOT_AUTHORITY, _DECIDE_AUTHORITY, _ENVELOPE_AUTHORITY], + "decision_reason": refusal_reason, + "trace_summary": { + "authority_evaluated": True, + "envelope_version": "local-v1", + "action_fingerprint": _action_fingerprint(normalized_payload), + "proposer_trace_hash_ignored": "trace_hash" + in normalized_payload["proposer"], + "execution_performed": False, + "path_normalized": False, + }, + "refusal_reason": refusal_reason, + } + ) + return evaluate_authority(normalized_payload) __all__ = [ diff --git a/tests/test_claude_tool_authority_demo.py b/tests/test_claude_tool_authority_demo.py index e4243064..013adb81 100644 --- a/tests/test_claude_tool_authority_demo.py +++ b/tests/test_claude_tool_authority_demo.py @@ -1,6 +1,7 @@ from __future__ import annotations import ast +import copy import hashlib import json from pathlib import Path @@ -96,6 +97,28 @@ def test_authorized_only_inside_envelope(): assert refused["refusal_reason"] == "protected_path" +@pytest.mark.parametrize( + ("target_path", "expected_reason"), + [ + ("workspace/demo_notes/../../core/authority.py", "path_traversal_detected"), + ("workspace/demo_notes/../demo_notes/allowed.md", "path_traversal_detected"), + ("/absolute/path.md", "absolute_target_path"), + ("workspace/demo_notes/subdir/../../../README.md", "path_traversal_detected"), + ], +) +def test_traversal_and_absolute_paths_are_refused(target_path: str, expected_reason: str): + payload = _fixture("authorized-low-risk-local-action.json")["arguments"] + original = copy.deepcopy(payload) + payload["action_request"]["target"]["path"] = target_path + response = authority.run_authority(payload) + assert response["status"] == "refused" + assert response["refusal_reason"] == expected_reason + assert response["decision_reason"] == expected_reason + assert response["trace_summary"]["authority_evaluated"] is True + assert original["action_request"]["target"]["path"] == "workspace/demo_notes/meeting-summary.txt" + assert payload["action_request"]["target"]["path"] == target_path + + def test_invalid_payload_fails_before_authority_evaluation(monkeypatch): def _explode(_payload: dict[str, object]) -> dict[str, object]: raise AssertionError("authority evaluation ran for invalid payload")