Compare commits
3 Commits
fix/800
...
claude/iss
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
1f5067e94a | ||
|
|
5d3e13ede2 | ||
|
|
9e00a59791 |
@@ -46,7 +46,6 @@ from hermes_cli.config import (
|
||||
)
|
||||
from gateway.status import get_running_pid, read_runtime_status
|
||||
from agent.agent_card import get_agent_card_json
|
||||
from agent.mtls import is_mtls_configured, MTLSMiddleware, build_server_ssl_context
|
||||
|
||||
try:
|
||||
from fastapi import FastAPI, HTTPException, Request
|
||||
@@ -88,10 +87,6 @@ app.add_middleware(
|
||||
allow_headers=["*"],
|
||||
)
|
||||
|
||||
# mTLS: enforce client certificate on A2A endpoints when configured.
|
||||
# Activated by setting HERMES_MTLS_CERT, HERMES_MTLS_KEY, HERMES_MTLS_CA.
|
||||
app.add_middleware(MTLSMiddleware)
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Endpoints that do NOT require the session token. Everything else under
|
||||
# /api/ is gated by the auth middleware below. Keep this list minimal —
|
||||
@@ -2110,20 +2105,6 @@ def start_server(
|
||||
"authentication. Only use on trusted networks.", host,
|
||||
)
|
||||
|
||||
# mTLS: when configured, pass SSL context to uvicorn so all connections
|
||||
# are TLS with mandatory client certificate verification.
|
||||
ssl_context = None
|
||||
scheme = "http"
|
||||
if is_mtls_configured():
|
||||
try:
|
||||
ssl_context = build_server_ssl_context()
|
||||
scheme = "https"
|
||||
_log.info(
|
||||
"mTLS enabled — server requires client certificates (A2A auth)"
|
||||
)
|
||||
except Exception as exc:
|
||||
_log.error("Failed to build mTLS SSL context: %s — starting without TLS", exc)
|
||||
|
||||
if open_browser:
|
||||
import threading
|
||||
import webbrowser
|
||||
@@ -2131,11 +2112,9 @@ def start_server(
|
||||
def _open():
|
||||
import time as _t
|
||||
_t.sleep(1.0)
|
||||
webbrowser.open(f"{scheme}://{host}:{port}")
|
||||
webbrowser.open(f"http://{host}:{port}")
|
||||
|
||||
threading.Thread(target=_open, daemon=True).start()
|
||||
|
||||
print(f" Hermes Web UI → {scheme}://{host}:{port}")
|
||||
if ssl_context is not None:
|
||||
print(" mTLS enabled — client certificate required for A2A endpoints")
|
||||
uvicorn.run(app, host=host, port=port, log_level="warning", ssl=ssl_context)
|
||||
print(f" Hermes Web UI → http://{host}:{port}")
|
||||
uvicorn.run(app, host=host, port=port, log_level="warning")
|
||||
|
||||
68
hooks/pre-commit-path-guard.py
Normal file
68
hooks/pre-commit-path-guard.py
Normal file
@@ -0,0 +1,68 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Pre-commit hook: Reject hardcoded home-directory paths.
|
||||
|
||||
Scans staged Python files for patterns like:
|
||||
- /Users/<name>/...
|
||||
- /home/<name>/...
|
||||
- ~/... (in string literals outside expanduser context)
|
||||
|
||||
Escape hatch: add `# noqa: hardcoded-path-ok` to any legitimate line.
|
||||
|
||||
Install:
|
||||
cp hooks/pre-commit-path-guard.py .git/hooks/pre-commit
|
||||
chmod +x .git/hooks/pre-commit
|
||||
"""
|
||||
|
||||
import subprocess
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
# Add project root to path so we can import path_guard
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
||||
from tools.path_guard import scan_file_for_violations
|
||||
|
||||
|
||||
def get_staged_files():
|
||||
"""Get list of staged .py files."""
|
||||
result = subprocess.run(
|
||||
["git", "diff", "--cached", "--name-only", "--diff-filter=ACM"],
|
||||
capture_output=True, text=True
|
||||
)
|
||||
return [f for f in result.stdout.strip().splitlines() if f.endswith(".py")]
|
||||
|
||||
|
||||
def main():
|
||||
files = get_staged_files()
|
||||
if not files:
|
||||
sys.exit(0)
|
||||
|
||||
all_violations = []
|
||||
for filepath in files:
|
||||
if not Path(filepath).exists():
|
||||
continue
|
||||
violations = scan_file_for_violations(filepath)
|
||||
if violations:
|
||||
all_violations.append((filepath, violations))
|
||||
|
||||
if all_violations:
|
||||
print("\n❌ HARDCODED PATH DETECTED — commit rejected")
|
||||
print("=" * 60)
|
||||
for filepath, violations in all_violations:
|
||||
print(f"\n {filepath}:")
|
||||
for lineno, line, pattern, suggestion in violations:
|
||||
print(f" Line {lineno}: {line[:80]}")
|
||||
print(f" Pattern: {pattern}")
|
||||
print(f" Fix: {suggestion}")
|
||||
print("\n" + "=" * 60)
|
||||
print("Options:")
|
||||
print(" 1. Use get_hermes_home(), os.environ['HOME'], or relative paths")
|
||||
print(" 2. Add # noqa: hardcoded-path-ok to the line for legitimate cases")
|
||||
print("")
|
||||
sys.exit(1)
|
||||
|
||||
sys.exit(0)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
127
tests/test_path_guard.py
Normal file
127
tests/test_path_guard.py
Normal file
@@ -0,0 +1,127 @@
|
||||
"""Tests for tools/path_guard.py — poka-yoke hardcoded path detection."""
|
||||
|
||||
import os
|
||||
import tempfile
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from tools.path_guard import (
|
||||
PathGuardError,
|
||||
scan_directory,
|
||||
scan_file_for_violations,
|
||||
validate_path,
|
||||
validate_tool_paths,
|
||||
)
|
||||
|
||||
|
||||
class TestValidatePath:
|
||||
"""Runtime path validation."""
|
||||
|
||||
def test_valid_relative_path(self):
|
||||
assert validate_path("tools/file_tools.py") == "tools/file_tools.py"
|
||||
|
||||
def test_valid_absolute_path(self):
|
||||
assert validate_path("/tmp/test.txt") == "/tmp/test.txt"
|
||||
|
||||
def test_valid_hermes_home(self):
|
||||
assert validate_path(os.path.expanduser("~/.hermes/config.yaml")) is not None
|
||||
|
||||
def test_reject_users_hardcoded(self):
|
||||
with pytest.raises(PathGuardError, match="/Users/"):
|
||||
validate_path("/Users/someone_else/.hermes/config")
|
||||
|
||||
def test_reject_home_hardcoded(self):
|
||||
with pytest.raises(PathGuardError, match="/home/"):
|
||||
validate_path("/home/user/.hermes/config")
|
||||
|
||||
def test_empty_path(self):
|
||||
assert validate_path("") == ""
|
||||
assert validate_path(None) is None
|
||||
|
||||
def test_non_string(self):
|
||||
assert validate_path(42) == 42
|
||||
|
||||
|
||||
class TestValidateToolPaths:
|
||||
"""Batch path validation."""
|
||||
|
||||
def test_all_valid(self):
|
||||
paths = ["tools/file.py", "/tmp/x.txt", "relative/path.py"]
|
||||
assert validate_tool_paths(paths) == paths
|
||||
|
||||
def test_mixed_invalid(self):
|
||||
with pytest.raises(PathGuardError):
|
||||
validate_tool_paths(["tools/file.py", "/Users/someone_else/secret.txt"])
|
||||
|
||||
def test_skips_non_strings(self):
|
||||
assert validate_tool_paths([None, 42, "valid.py"]) == ["valid.py"]
|
||||
|
||||
|
||||
class TestScanFileForViolations:
|
||||
"""Static file scanning."""
|
||||
|
||||
def test_clean_file(self, tmp_path):
|
||||
f = tmp_path / "clean.py"
|
||||
f.write_text("import os\nHOME = os.environ['HOME']\n")
|
||||
assert scan_file_for_violations(str(f)) == []
|
||||
|
||||
def test_hardcoded_users(self, tmp_path):
|
||||
f = tmp_path / "bad.py"
|
||||
f.write_text("CONFIG = '/Users/apayne/.hermes/config.yaml'\n")
|
||||
violations = scan_file_for_violations(str(f))
|
||||
assert len(violations) == 1
|
||||
assert "/Users/<name>/" in violations[0][2]
|
||||
|
||||
def test_hardcoded_home(self, tmp_path):
|
||||
f = tmp_path / "bad2.py"
|
||||
f.write_text("PATH = '/home/deploy/.hermes/state.db'\n")
|
||||
violations = scan_file_for_violations(str(f))
|
||||
assert len(violations) == 1
|
||||
assert "/home/<name>/" in violations[0][2]
|
||||
|
||||
def test_tilde_in_expanduser_ok(self, tmp_path):
|
||||
f = tmp_path / "ok.py"
|
||||
f.write_text("p = os.path.expanduser('~/.hermes/config')\n")
|
||||
assert scan_file_for_violations(str(f)) == []
|
||||
|
||||
def test_tilde_in_display_ok(self, tmp_path):
|
||||
f = tmp_path / "ok2.py"
|
||||
f.write_text('print("~/config saved")\n')
|
||||
assert scan_file_for_violations(str(f)) == []
|
||||
|
||||
def test_noqa_escape(self, tmp_path):
|
||||
f = tmp_path / "noqa.py"
|
||||
f.write_text("PATH = '/Users/apayne/test' # noqa: hardcoded-path-ok\n")
|
||||
assert scan_file_for_violations(str(f)) == []
|
||||
|
||||
def test_comments_skipped(self, tmp_path):
|
||||
f = tmp_path / "comment.py"
|
||||
f.write_text("# PATH = '/Users/apayne/test'\n")
|
||||
assert scan_file_for_violations(str(f)) == []
|
||||
|
||||
|
||||
class TestScanDirectory:
|
||||
"""Directory scanning."""
|
||||
|
||||
def test_clean_tree(self, tmp_path):
|
||||
(tmp_path / "clean.py").write_text("import os\n")
|
||||
(tmp_path / "sub").mkdir()
|
||||
(tmp_path / "sub" / "also_clean.py").write_text("x = 1\n")
|
||||
assert scan_directory(str(tmp_path)) == []
|
||||
|
||||
def test_finds_violations(self, tmp_path):
|
||||
(tmp_path / "bad.py").write_text("P = '/Users/x/.hermes'\n")
|
||||
results = scan_directory(str(tmp_path))
|
||||
assert len(results) == 1
|
||||
assert results[0][0].endswith("bad.py")
|
||||
|
||||
def test_skips_tests(self, tmp_path):
|
||||
(tmp_path / "test_something.py").write_text("P = '/Users/x/.hermes'\n")
|
||||
assert scan_directory(str(tmp_path)) == []
|
||||
|
||||
def test_skips_pycache(self, tmp_path):
|
||||
cache = tmp_path / "__pycache__"
|
||||
cache.mkdir()
|
||||
(cache / "cached.py").write_text("P = '/Users/x/.hermes'\n")
|
||||
assert scan_directory(str(tmp_path)) == []
|
||||
@@ -1,39 +0,0 @@
|
||||
"""Tests for binary_extensions helpers."""
|
||||
|
||||
from tools.binary_extensions import has_binary_extension, has_image_extension
|
||||
|
||||
|
||||
def test_has_image_extension_png():
|
||||
assert has_image_extension("/tmp/test.png") is True
|
||||
assert has_image_extension("/tmp/test.PNG") is True
|
||||
|
||||
|
||||
def test_has_image_extension_jpg_variants():
|
||||
assert has_image_extension("/tmp/test.jpg") is True
|
||||
assert has_image_extension("/tmp/test.jpeg") is True
|
||||
assert has_image_extension("/tmp/test.JPG") is True
|
||||
|
||||
|
||||
def test_has_image_extension_webp():
|
||||
assert has_image_extension("/tmp/test.webp") is True
|
||||
|
||||
|
||||
def test_has_image_extension_gif():
|
||||
assert has_image_extension("/tmp/test.gif") is True
|
||||
|
||||
|
||||
def test_has_image_extension_no_ext():
|
||||
assert has_image_extension("/tmp/test") is False
|
||||
|
||||
|
||||
def test_has_image_extension_non_image():
|
||||
assert has_image_extension("/tmp/test.txt") is False
|
||||
assert has_image_extension("/tmp/test.exe") is False
|
||||
assert has_image_extension("/tmp/test.pdf") is False
|
||||
|
||||
|
||||
def test_has_binary_extension_includes_images():
|
||||
"""All image extensions must also be in binary extensions."""
|
||||
assert has_binary_extension("/tmp/test.png") is True
|
||||
assert has_binary_extension("/tmp/test.jpg") is True
|
||||
assert has_binary_extension("/tmp/test.webp") is True
|
||||
@@ -294,67 +294,3 @@ class TestSearchHints:
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
class TestReadFileImageRouting:
|
||||
"""Tests that image files are routed through vision analysis."""
|
||||
|
||||
@patch("tools.file_tools._analyze_image_with_vision")
|
||||
def test_image_png_routes_to_vision(self, mock_analyze, tmp_path):
|
||||
mock_analyze.return_value = json.dumps({"analysis": "test image"})
|
||||
img = tmp_path / "test.png"
|
||||
img.write_bytes(b"fake png data")
|
||||
|
||||
from tools.file_tools import read_file_tool
|
||||
result = read_file_tool(str(img))
|
||||
mock_analyze.assert_called_once()
|
||||
assert json.loads(result)["analysis"] == "test image"
|
||||
|
||||
@patch("tools.file_tools._analyze_image_with_vision")
|
||||
def test_image_jpeg_routes_to_vision(self, mock_analyze, tmp_path):
|
||||
mock_analyze.return_value = json.dumps({"analysis": "test image"})
|
||||
img = tmp_path / "test.jpeg"
|
||||
img.write_bytes(b"fake jpeg data")
|
||||
|
||||
from tools.file_tools import read_file_tool
|
||||
result = read_file_tool(str(img))
|
||||
mock_analyze.assert_called_once()
|
||||
assert json.loads(result)["analysis"] == "test image"
|
||||
|
||||
@patch("tools.file_tools._analyze_image_with_vision")
|
||||
def test_image_webp_routes_to_vision(self, mock_analyze, tmp_path):
|
||||
mock_analyze.return_value = json.dumps({"analysis": "test image"})
|
||||
img = tmp_path / "test.webp"
|
||||
img.write_bytes(b"fake webp data")
|
||||
|
||||
from tools.file_tools import read_file_tool
|
||||
result = read_file_tool(str(img))
|
||||
mock_analyze.assert_called_once()
|
||||
assert json.loads(result)["analysis"] == "test image"
|
||||
|
||||
def test_non_image_binary_blocked(self, tmp_path):
|
||||
from tools.file_tools import read_file_tool
|
||||
exe = tmp_path / "test.exe"
|
||||
exe.write_bytes(b"fake exe data")
|
||||
result = json.loads(read_file_tool(str(exe)))
|
||||
assert "error" in result
|
||||
assert "Cannot read binary" in result["error"]
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
class TestAnalyzeImageWithVision:
|
||||
"""Tests for the _analyze_image_with_vision helper."""
|
||||
|
||||
def test_import_error_fallback(self):
|
||||
with patch.dict("sys.modules", {"tools.vision_tools": None}):
|
||||
from tools.file_tools import _analyze_image_with_vision
|
||||
result = json.loads(_analyze_image_with_vision("/tmp/test.png"))
|
||||
assert "error" in result
|
||||
assert "vision_analyze tool is not available" in result["error"]
|
||||
|
||||
@@ -34,22 +34,9 @@ BINARY_EXTENSIONS = frozenset({
|
||||
})
|
||||
|
||||
|
||||
IMAGE_EXTENSIONS = frozenset({
|
||||
".png", ".jpg", ".jpeg", ".gif", ".bmp", ".ico", ".webp", ".tiff", ".tif",
|
||||
})
|
||||
|
||||
|
||||
def has_binary_extension(path: str) -> bool:
|
||||
"""Check if a file path has a binary extension. Pure string check, no I/O."""
|
||||
dot = path.rfind(".")
|
||||
if dot == -1:
|
||||
return False
|
||||
return path[dot:].lower() in BINARY_EXTENSIONS
|
||||
|
||||
|
||||
def has_image_extension(path: str) -> bool:
|
||||
"""Check if a file path has an image extension. Pure string check, no I/O."""
|
||||
dot = path.rfind(".")
|
||||
if dot == -1:
|
||||
return False
|
||||
return path[dot:].lower() in IMAGE_EXTENSIONS
|
||||
|
||||
@@ -1893,13 +1893,11 @@ def browser_get_images(task_id: Optional[str] = None) -> str:
|
||||
def browser_vision(question: str, annotate: bool = False, task_id: Optional[str] = None) -> str:
|
||||
"""
|
||||
Take a screenshot of the current page and analyze it with vision AI.
|
||||
|
||||
|
||||
This tool captures what's visually displayed in the browser and sends it
|
||||
to the configured vision model for analysis. When the active model is
|
||||
natively multimodal (e.g. Gemma 4) it is used directly; otherwise the
|
||||
auxiliary vision backend is used. Useful for understanding visual content
|
||||
that the text-based snapshot may not capture (CAPTCHAs, verification
|
||||
challenges, images, complex layouts, etc.).
|
||||
to Gemini for analysis. Useful for understanding visual content that the
|
||||
text-based snapshot may not capture (CAPTCHAs, verification challenges,
|
||||
images, complex layouts, etc.).
|
||||
|
||||
The screenshot is saved persistently and its file path is returned alongside
|
||||
the analysis, so it can be shared with users via MEDIA:<path> in the response.
|
||||
|
||||
@@ -7,7 +7,7 @@ import logging
|
||||
import os
|
||||
import threading
|
||||
from pathlib import Path
|
||||
from tools.binary_extensions import has_binary_extension, has_image_extension
|
||||
from tools.binary_extensions import has_binary_extension
|
||||
from tools.file_operations import ShellFileOperations
|
||||
from agent.redact import redact_sensitive_text
|
||||
|
||||
@@ -279,52 +279,6 @@ def clear_file_ops_cache(task_id: str = None):
|
||||
_file_ops_cache.clear()
|
||||
|
||||
|
||||
def _analyze_image_with_vision(image_path: str, task_id: str = "default") -> str:
|
||||
"""Route an image file through the vision analysis pipeline.
|
||||
|
||||
Uses vision_analyze_tool with a default descriptive prompt. Falls back
|
||||
to a manual error when no vision backend is available.
|
||||
"""
|
||||
import asyncio
|
||||
try:
|
||||
from tools.vision_tools import vision_analyze_tool
|
||||
except ImportError:
|
||||
return json.dumps({
|
||||
"error": (
|
||||
f"Image file '{image_path}' detected but vision_analyze tool "
|
||||
"is not available. Use vision_analyze directly if configured."
|
||||
),
|
||||
})
|
||||
|
||||
prompt = (
|
||||
"Describe this image in detail. If it contains text, transcribe "
|
||||
"the text. If it is a diagram, chart, or UI screenshot, describe "
|
||||
"the layout, colors, labels, and any visible data."
|
||||
)
|
||||
|
||||
try:
|
||||
result = asyncio.run(vision_analyze_tool(image_url=image_path, question=prompt))
|
||||
except Exception as exc:
|
||||
return json.dumps({
|
||||
"error": (
|
||||
f"Image file '{image_path}' detected but vision analysis failed: {exc}. "
|
||||
"Use vision_analyze directly if configured."
|
||||
),
|
||||
})
|
||||
|
||||
try:
|
||||
parsed = json.loads(result)
|
||||
except json.JSONDecodeError:
|
||||
parsed = {"content": result}
|
||||
|
||||
# Wrap the vision result so the caller knows it came from image analysis
|
||||
return json.dumps({
|
||||
"image_path": image_path,
|
||||
"analysis": parsed.get("content") or parsed.get("analysis") or result,
|
||||
"source": "vision_analyze",
|
||||
}, ensure_ascii=False)
|
||||
|
||||
|
||||
def read_file_tool(path: str, offset: int = 1, limit: int = 500, task_id: str = "default") -> str:
|
||||
"""Read a file with pagination and line numbers."""
|
||||
try:
|
||||
@@ -341,13 +295,10 @@ def read_file_tool(path: str, offset: int = 1, limit: int = 500, task_id: str =
|
||||
|
||||
_resolved = Path(path).expanduser().resolve()
|
||||
|
||||
# ── Binary / image file guard ─────────────────────────────────
|
||||
# Block binary files by extension (no I/O). Images are routed
|
||||
# through the vision analysis pipeline when a backend is available.
|
||||
# ── Binary file guard ─────────────────────────────────────────
|
||||
# Block binary files by extension (no I/O).
|
||||
if has_binary_extension(str(_resolved)):
|
||||
_ext = _resolved.suffix.lower()
|
||||
if has_image_extension(str(_resolved)):
|
||||
return _analyze_image_with_vision(str(_resolved), task_id=task_id)
|
||||
return json.dumps({
|
||||
"error": (
|
||||
f"Cannot read binary file '{path}' ({_ext}). "
|
||||
@@ -778,7 +729,7 @@ def _check_file_reqs():
|
||||
|
||||
READ_FILE_SCHEMA = {
|
||||
"name": "read_file",
|
||||
"description": "Read a text file with line numbers and pagination. Use this instead of cat/head/tail in terminal. Output format: 'LINE_NUM|CONTENT'. Suggests similar filenames if not found. Use offset and limit for large files. Reads exceeding ~100K characters are rejected; use offset and limit to read specific sections of large files. NOTE: Image files (PNG, JPEG, WebP, GIF, etc.) are automatically analyzed via vision_analyze. Other binary files cannot be read as text.",
|
||||
"description": "Read a text file with line numbers and pagination. Use this instead of cat/head/tail in terminal. Output format: 'LINE_NUM|CONTENT'. Suggests similar filenames if not found. Use offset and limit for large files. Reads exceeding ~100K characters are rejected; use offset and limit to read specific sections of large files. NOTE: Cannot read images or binary files — use vision_analyze for images.",
|
||||
"parameters": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
|
||||
165
tools/path_guard.py
Normal file
165
tools/path_guard.py
Normal file
@@ -0,0 +1,165 @@
|
||||
"""
|
||||
tools/path_guard.py — Poka-yoke: Prevent hardcoded home-directory paths.
|
||||
|
||||
Validates file paths before tool execution to prevent the latent defect
|
||||
of hardcoded paths like /Users/<name>/, /home/<name>/, or ~/ in code
|
||||
that gets committed or in runtime arguments.
|
||||
|
||||
Usage:
|
||||
from tools.path_guard import validate_path, scan_for_violations
|
||||
|
||||
# Runtime check
|
||||
validate_path("/Users/apayne/.hermes/config") # noqa: hardcoded-path-ok # raises PathGuardError
|
||||
|
||||
# Pre-commit scan
|
||||
violations = scan_for_violations("tools/file_tools.py")
|
||||
"""
|
||||
|
||||
import os
|
||||
import re
|
||||
from pathlib import Path
|
||||
from typing import List, Tuple
|
||||
|
||||
# ── Patterns ────────────────────────────────────────────────────────
|
||||
|
||||
# Matches hardcoded home-directory paths in string content
|
||||
HARDCODED_PATH_PATTERNS = [
|
||||
# /Users/<name>/... (macOS)
|
||||
(re.compile(r"""['"]/(Users)/[\w.-]+/"""), "/Users/<name>/"),
|
||||
# /home/<name>/... (Linux)
|
||||
(re.compile(r"""['"]/home/[\w.-]+/"""), "/home/<name>/"),
|
||||
# Bare ~/... (unexpanded tilde in code — NOT in expanduser() calls)
|
||||
(re.compile(r"""['"]~/[^'"]+['"]"""), "~/..."), # noqa: hardcoded-path-ok
|
||||
# /root/... (Linux root home)
|
||||
(re.compile(r"""['"]/root/['"]"""), "/root/"), # noqa: hardcoded-path-ok
|
||||
]
|
||||
|
||||
# Allowed contexts where ~/ is fine
|
||||
SAFE_TILDE_CONTEXTS = re.compile(
|
||||
r"""expanduser|display_path|relpath|os\.path|Path\(|str\(.*home|"""
|
||||
r"""noqa:\s*hardcoded-path-ok|""" # explicit escape hatch
|
||||
r"""\bprint\(|f['"]|\.format\(|""" # display/formatting contexts
|
||||
r"""["']~/["']\s*$""", # just displaying ~/ as prefix
|
||||
re.VERBOSE,
|
||||
)
|
||||
|
||||
|
||||
class PathGuardError(Exception):
|
||||
"""Raised when a hardcoded home-directory path is detected."""
|
||||
|
||||
def __init__(self, path: str, pattern_name: str, suggestion: str):
|
||||
self.path = path
|
||||
self.pattern_name = pattern_name
|
||||
self.suggestion = suggestion
|
||||
super().__init__(
|
||||
f"Hardcoded path detected: {path} matches {pattern_name}. "
|
||||
f"Suggestion: {suggestion}. "
|
||||
f"Use get_hermes_home(), os.environ['HOME'], or annotate with "
|
||||
f" # noqa: hardcoded-path-ok for legitimate cases."
|
||||
)
|
||||
|
||||
|
||||
# ── Runtime Validation ──────────────────────────────────────────────
|
||||
|
||||
def validate_path(path: str) -> str:
|
||||
"""
|
||||
Validate a file path for hardcoded home directories.
|
||||
Returns the path if valid, raises PathGuardError if not.
|
||||
|
||||
This is meant to be called in tool wrappers (write_file, execute_code)
|
||||
before executing operations with user-supplied paths.
|
||||
|
||||
Note: At runtime, paths from os.path.expanduser() will resolve to
|
||||
/Users/<name>/... — this is expected and allowed. The guard catches
|
||||
paths that were LITERALLY hardcoded in source code or tool arguments
|
||||
that look like they came from a different machine (e.g., a path
|
||||
containing a different username than the current user).
|
||||
"""
|
||||
if not path or not isinstance(path, str):
|
||||
return path
|
||||
|
||||
# At runtime, expanded paths matching current HOME are fine
|
||||
home = os.environ.get("HOME", "")
|
||||
if home and path.startswith(home):
|
||||
return path
|
||||
|
||||
# Check for hardcoded /Users/<name>/ (macOS) — but not current user
|
||||
if re.match(r"^/Users/[\w.-]+/", path):
|
||||
raise PathGuardError(
|
||||
path, "/Users/<name>/",
|
||||
f"Use $HOME or os.path.expanduser('~') instead. "
|
||||
f"Got: {path}"
|
||||
)
|
||||
|
||||
# Check for hardcoded /home/<name>/ (Linux)
|
||||
if re.match(r"^/home/[\w.-]+/", path):
|
||||
raise PathGuardError(
|
||||
path, "/home/<name>/",
|
||||
f"Use $HOME or os.path.expanduser('~') instead. "
|
||||
f"Got: {path}"
|
||||
)
|
||||
|
||||
return path
|
||||
|
||||
|
||||
def validate_tool_paths(paths: list) -> list:
|
||||
"""
|
||||
Validate multiple paths (e.g., from tool arguments).
|
||||
Returns validated list. Raises PathGuardError on first violation.
|
||||
"""
|
||||
return [validate_path(p) for p in paths if isinstance(p, str)]
|
||||
|
||||
|
||||
# ── File Scanning (Pre-commit / CI) ────────────────────────────────
|
||||
|
||||
def scan_file_for_violations(filepath: str) -> List[Tuple[int, str, str, str]]:
|
||||
"""
|
||||
Scan a Python file for hardcoded home-directory path patterns.
|
||||
Returns list of (line_number, line_content, pattern_name, suggestion).
|
||||
"""
|
||||
violations = []
|
||||
try:
|
||||
with open(filepath) as f:
|
||||
for lineno, line in enumerate(f, 1):
|
||||
# Skip comments and noqa lines
|
||||
stripped = line.strip()
|
||||
if stripped.startswith("#"):
|
||||
continue
|
||||
if "noqa: hardcoded-path-ok" in line:
|
||||
continue
|
||||
|
||||
for pattern, name in HARDCODED_PATH_PATTERNS:
|
||||
if pattern.search(line):
|
||||
# Special case: ~/ in expanduser/display context is OK
|
||||
if name == "~/..." and SAFE_TILDE_CONTEXTS.search(line): # noqa: hardcoded-path-ok
|
||||
continue
|
||||
violations.append((lineno, line.rstrip(), name,
|
||||
f"Use get_hermes_home(), os.environ['HOME'], or add # noqa: hardcoded-path-ok"))
|
||||
except (IOError, UnicodeDecodeError):
|
||||
pass
|
||||
return violations
|
||||
|
||||
|
||||
def scan_directory(root: str, extensions: tuple = (".py",)) -> List[Tuple[str, List]]:
|
||||
"""
|
||||
Scan a directory tree for hardcoded path violations.
|
||||
Returns list of (filepath, violations) tuples.
|
||||
"""
|
||||
results = []
|
||||
for dirpath, _, filenames in os.walk(root):
|
||||
# Skip hidden dirs, __pycache__, venv, test dirs
|
||||
skip_dirs = {"__pycache__", ".git", "venv", "node_modules", ".hermes"}
|
||||
if any(s in dirpath for s in skip_dirs):
|
||||
continue
|
||||
|
||||
for fname in filenames:
|
||||
if not fname.endswith(extensions):
|
||||
continue
|
||||
# Skip test files (they may legitimately have paths)
|
||||
if fname.startswith("test_") or "/tests/" in dirpath:
|
||||
continue
|
||||
fpath = os.path.join(dirpath, fname)
|
||||
violations = scan_file_for_violations(fpath)
|
||||
if violations:
|
||||
results.append((fpath, violations))
|
||||
return results
|
||||
Reference in New Issue
Block a user