🔧 Initial dev copy from live
This commit is contained in:
Rose
310
tests/test_security_redaction.py
Normal file
310
tests/test_security_redaction.py
Normal file
@@ -0,0 +1,310 @@
|
||||
"""
|
||||
Security tests: credential redaction in API responses.
|
||||
|
||||
Verifies that credentials (GitHub PATs, API keys, etc.) are masked in:
|
||||
- GET /api/session (messages and tool_calls)
|
||||
- GET /api/memory (MEMORY.md and USER.md content)
|
||||
- GET /api/session/export (downloaded JSON)
|
||||
- SSE done event (session payload in stream)
|
||||
|
||||
Tests run against the isolated test test_server on port 8788.
|
||||
"""
|
||||
|
||||
import json
|
||||
import pathlib
|
||||
import sys
|
||||
import urllib.request
|
||||
import urllib.error
|
||||
import pytest
|
||||
|
||||
sys.path.insert(0, str(pathlib.Path(__file__).parent.parent.parent))
|
||||
|
||||
|
||||
def _server_is_up(port: int = 8788) -> bool:
|
||||
"""Return True if the test server is accepting connections."""
|
||||
try:
|
||||
urllib.request.urlopen(f"http://127.0.0.1:{port}/health", timeout=2)
|
||||
return True
|
||||
except Exception:
|
||||
return False
|
||||
|
||||
|
||||
# _needs_server: these tests require the conftest test_server fixture (port 8788).
|
||||
# The skipif is evaluated lazily via the fixture, not at collection time.
|
||||
_needs_server = pytest.mark.usefixtures("test_server")
|
||||
|
||||
from tests._pytest_port import BASE
|
||||
|
||||
# Sample credentials that should be masked in every API response
|
||||
_FAKE_GITHUB_PAT = "ghp_TestFakeCredential1234567890ab"
|
||||
_FAKE_SK_KEY = "sk-TestFakeOpenAIKey1234567890abcdef"
|
||||
_FAKE_HF_TOKEN = "hf_TestFakeHuggingFaceToken12345"
|
||||
_FAKE_AWS_KEY = "AKIATESTFAKEKEY12345"
|
||||
|
||||
|
||||
# ── HTTP helpers ──────────────────────────────────────────────────────────────
|
||||
|
||||
def _get(path):
|
||||
with urllib.request.urlopen(BASE + path, timeout=10) as r:
|
||||
return json.loads(r.read())
|
||||
|
||||
|
||||
def _post(path, body=None):
|
||||
data = json.dumps(body or {}).encode()
|
||||
req = urllib.request.Request(
|
||||
BASE + path, data=data,
|
||||
headers={"Content-Type": "application/json"},
|
||||
)
|
||||
try:
|
||||
with urllib.request.urlopen(req, timeout=10) as r:
|
||||
return json.loads(r.read()), r.status
|
||||
except urllib.error.HTTPError as e:
|
||||
return json.loads(e.read()), e.code
|
||||
|
||||
|
||||
def _get_raw(path):
|
||||
"""Return raw bytes (used for export endpoint)."""
|
||||
with urllib.request.urlopen(BASE + path, timeout=10) as r:
|
||||
return r.read()
|
||||
|
||||
|
||||
def _assert_no_plaintext_credentials(text: str, label: str = ""):
|
||||
"""Assert that none of the fake credential strings appear in text."""
|
||||
for cred in (_FAKE_GITHUB_PAT, _FAKE_SK_KEY, _FAKE_HF_TOKEN, _FAKE_AWS_KEY):
|
||||
assert cred not in text, (
|
||||
f"{label}: credential '{cred[:12]}...' found in plaintext. "
|
||||
"Redaction is not working."
|
||||
)
|
||||
|
||||
|
||||
# ── helpers.py unit tests (import-level, no test_server needed) ───────────────────
|
||||
|
||||
def test_redact_value_str():
|
||||
"""_redact_value masks a plaintext GitHub PAT in a string."""
|
||||
from api.helpers import _redact_value
|
||||
result = _redact_value(f"my token is {_FAKE_GITHUB_PAT} bye")
|
||||
assert _FAKE_GITHUB_PAT not in result
|
||||
assert "ghp_Te" in result # prefix preserved
|
||||
|
||||
|
||||
def test_redact_value_dict():
|
||||
"""_redact_value recurses into dicts."""
|
||||
from api.helpers import _redact_value
|
||||
d = {"content": f"key={_FAKE_SK_KEY}", "role": "user"}
|
||||
result = _redact_value(d)
|
||||
assert _FAKE_SK_KEY not in result["content"]
|
||||
assert result["role"] == "user" # innocent values untouched
|
||||
|
||||
|
||||
def test_redact_value_list():
|
||||
"""_redact_value recurses into lists."""
|
||||
from api.helpers import _redact_value
|
||||
lst = [{"content": _FAKE_GITHUB_PAT}, {"content": "safe text"}]
|
||||
result = _redact_value(lst)
|
||||
assert _FAKE_GITHUB_PAT not in result[0]["content"]
|
||||
assert result[1]["content"] == "safe text"
|
||||
|
||||
|
||||
def test_redact_session_data_messages():
|
||||
"""redact_session_data masks credentials in messages[]."""
|
||||
from api.helpers import redact_session_data
|
||||
session = {
|
||||
"session_id": "abc123",
|
||||
"title": f"my token {_FAKE_GITHUB_PAT}",
|
||||
"messages": [
|
||||
{"role": "user", "content": f"token: {_FAKE_GITHUB_PAT}"},
|
||||
{"role": "assistant", "content": "sure"},
|
||||
],
|
||||
"tool_calls": [
|
||||
{"name": "terminal", "args": {"command": f"gh auth login --token {_FAKE_GITHUB_PAT}"},
|
||||
"snippet": "ok"},
|
||||
],
|
||||
}
|
||||
result = redact_session_data(session)
|
||||
dump = json.dumps(result)
|
||||
_assert_no_plaintext_credentials(dump, "redact_session_data")
|
||||
# Safe fields remain intact
|
||||
assert result["session_id"] == "abc123"
|
||||
assert result["messages"][1]["content"] == "sure"
|
||||
|
||||
|
||||
def test_redact_session_data_multiple_cred_types():
|
||||
"""redact_session_data handles sk-, ghp_, hf_, and AKIA keys."""
|
||||
from api.helpers import redact_session_data
|
||||
session = {
|
||||
"title": "test",
|
||||
"messages": [{"role": "user", "content": (
|
||||
f"openai={_FAKE_SK_KEY} "
|
||||
f"github={_FAKE_GITHUB_PAT} "
|
||||
f"hf={_FAKE_HF_TOKEN} "
|
||||
f"aws={_FAKE_AWS_KEY}"
|
||||
)}],
|
||||
"tool_calls": [],
|
||||
}
|
||||
result = redact_session_data(session)
|
||||
dump = json.dumps(result)
|
||||
_assert_no_plaintext_credentials(dump, "multi-type redaction")
|
||||
|
||||
|
||||
def test_redact_session_data_non_sensitive_unchanged():
|
||||
"""redact_session_data does not corrupt innocent content."""
|
||||
from api.helpers import redact_session_data
|
||||
session = {
|
||||
"title": "Hello world",
|
||||
"messages": [{"role": "user", "content": "What is 2+2?"}],
|
||||
"tool_calls": [{"name": "terminal", "snippet": "4"}],
|
||||
}
|
||||
result = redact_session_data(session)
|
||||
assert result["title"] == "Hello world"
|
||||
assert result["messages"][0]["content"] == "What is 2+2?"
|
||||
assert result["tool_calls"][0]["snippet"] == "4"
|
||||
|
||||
|
||||
# ── API-level tests (require running test server started by conftest.py) ─────
|
||||
# Run via `start.sh && pytest tests/test_security_redaction.py -v`
|
||||
|
||||
def _create_session_with_credentials() -> str:
|
||||
"""Write a session file with credential-containing messages directly to disk.
|
||||
|
||||
Bypasses the server's in-memory cache so the GET endpoint is forced to read
|
||||
from disk, exercising the redaction code path on load.
|
||||
Uses TEST_STATE_DIR from conftest.py (the isolated test server state directory).
|
||||
"""
|
||||
import time, uuid
|
||||
try:
|
||||
from conftest import TEST_STATE_DIR
|
||||
sessions_dir = TEST_STATE_DIR / "sessions"
|
||||
except ImportError:
|
||||
from api.config import SESSION_DIR as sessions_dir
|
||||
sessions_dir = pathlib.Path(sessions_dir)
|
||||
sessions_dir.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
# Use a unique session ID that is NOT in the server's LRU cache
|
||||
sid = "sec_test_" + uuid.uuid4().hex[:8]
|
||||
now = time.time()
|
||||
session_file = sessions_dir / f"{sid}.json"
|
||||
session_file.write_text(json.dumps({
|
||||
"session_id": sid,
|
||||
"title": f"session with {_FAKE_GITHUB_PAT}",
|
||||
"workspace": "/tmp",
|
||||
"model": "test",
|
||||
"created_at": now,
|
||||
"updated_at": now,
|
||||
"pinned": False, "archived": False, "project_id": None,
|
||||
"profile": "default", "input_tokens": 0, "output_tokens": 0,
|
||||
"estimated_cost": None, "personality": None,
|
||||
"messages": [
|
||||
{"role": "user", "content": f"my PAT is {_FAKE_GITHUB_PAT}"},
|
||||
{"role": "assistant", "content": f"sk key is {_FAKE_SK_KEY}"},
|
||||
{"role": "tool", "content": "result ok", "name": "terminal"},
|
||||
],
|
||||
"tool_calls": [
|
||||
{"name": "terminal",
|
||||
"args": {"command": f"gh auth login --token {_FAKE_GITHUB_PAT}"},
|
||||
"snippet": "blocked"}
|
||||
],
|
||||
}))
|
||||
return sid
|
||||
|
||||
|
||||
def test_api_session_redacts_messages():
|
||||
"""GET /api/session route must call redact_session_data() before returning."""
|
||||
import inspect
|
||||
import api.routes as routes
|
||||
src = inspect.getsource(routes.handle_get)
|
||||
# Verify redact_session_data is applied to the session payload
|
||||
assert "redact_session_data" in src, (
|
||||
"api/routes.py handle_get must call redact_session_data() on /api/session response"
|
||||
)
|
||||
|
||||
|
||||
def test_api_session_redacts_title():
|
||||
"""redact_session_data must redact credentials from session title field."""
|
||||
from api.helpers import redact_session_data
|
||||
session = {
|
||||
"session_id": "abc123",
|
||||
"title": f"session with {_FAKE_GITHUB_PAT}",
|
||||
"messages": [],
|
||||
"tool_calls": [],
|
||||
}
|
||||
result = redact_session_data(session)
|
||||
assert _FAKE_GITHUB_PAT not in result["title"], (
|
||||
f"redact_session_data must mask credentials in title field"
|
||||
)
|
||||
assert result["session_id"] == "abc123" # safe fields preserved
|
||||
|
||||
|
||||
@_needs_server
|
||||
def test_api_sessions_list_redacts_titles(test_server):
|
||||
"""GET /api/sessions must not return session titles containing credentials."""
|
||||
_create_session_with_credentials()
|
||||
data = _get("/api/sessions")
|
||||
dump = json.dumps(data)
|
||||
_assert_no_plaintext_credentials(dump, "GET /api/sessions titles")
|
||||
|
||||
|
||||
def test_api_session_export_redacts():
|
||||
"""GET /api/session/export must call redact_session_data() in _handle_session_export."""
|
||||
import inspect
|
||||
import api.routes as routes
|
||||
# The export handler is a separate function (_handle_session_export)
|
||||
src = inspect.getsource(routes._handle_session_export)
|
||||
assert "redact_session_data" in src, (
|
||||
"_handle_session_export must call redact_session_data() before serving download"
|
||||
)
|
||||
|
||||
|
||||
@_needs_server
|
||||
def test_api_memory_redacts_via_write_read(test_server):
|
||||
"""Credential written to MEMORY.md must be masked in GET /api/memory response."""
|
||||
original = _get("/api/memory").get("memory", "")
|
||||
|
||||
cred_content = f"GitHub PAT: {_FAKE_GITHUB_PAT}\nNormal note: hello world"
|
||||
data, status = _post("/api/memory/write", {"section": "memory", "content": cred_content})
|
||||
assert status == 200, f"memory/write failed: {data}"
|
||||
|
||||
try:
|
||||
read_back = _get("/api/memory")
|
||||
dump = json.dumps(read_back)
|
||||
_assert_no_plaintext_credentials(dump, "GET /api/memory")
|
||||
assert "hello world" in read_back["memory"] # non-sensitive content preserved
|
||||
finally:
|
||||
_post("/api/memory/write", {"section": "memory", "content": original})
|
||||
|
||||
|
||||
# ── startup: fix_credential_permissions ──────────────────────────────────────
|
||||
|
||||
def test_fix_credential_permissions_corrects_loose_files(tmp_path, monkeypatch):
|
||||
"""fix_credential_permissions() tightens group/other read bits."""
|
||||
import os
|
||||
from api.startup import fix_credential_permissions
|
||||
|
||||
env_file = tmp_path / ".env"
|
||||
env_file.write_text("SECRET=abc")
|
||||
env_file.chmod(0o644) # world-readable -- should be fixed
|
||||
|
||||
google_file = tmp_path / "google_token.json"
|
||||
google_file.write_text("{}")
|
||||
google_file.chmod(0o664) # group-readable -- should be fixed
|
||||
|
||||
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
|
||||
fix_credential_permissions()
|
||||
|
||||
import stat
|
||||
assert stat.S_IMODE(env_file.stat().st_mode) == 0o600, ".env not fixed to 600"
|
||||
assert stat.S_IMODE(google_file.stat().st_mode) == 0o600, "google_token.json not fixed to 600"
|
||||
|
||||
|
||||
def test_fix_credential_permissions_skips_correct_files(tmp_path, monkeypatch):
|
||||
"""fix_credential_permissions() does not alter already-strict files."""
|
||||
env_file = tmp_path / ".env"
|
||||
env_file.write_text("SECRET=abc")
|
||||
env_file.chmod(0o600)
|
||||
|
||||
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
|
||||
|
||||
from api.startup import fix_credential_permissions
|
||||
fix_credential_permissions()
|
||||
|
||||
import stat
|
||||
assert stat.S_IMODE(env_file.stat().st_mode) == 0o600
|
||||
Reference in New Issue
Block a user