Gan, Jimmy
dac163d88c
Issue #2: Unprotected Chat Summary Trigger (CRITICAL) - Add authentication and admin authorization to /trigger endpoint - Add path validation to prevent directory traversal - Block access to system directories (/etc, /root, /sys, /proc, /boot) - Add tests for unauthorized access and invalid paths Issue #3: Exception Information Disclosure (HIGH) - Replace raw exception messages with generic errors - Log full exception details server-side with logger.exception() - Affected files: auth.py, files.py, passkey.py, opc_agents.py - Prevents information leakage about system architecture Security Impact: - Prevents unauthenticated file system writes - Reduces reconnaissance opportunities for attackers - Maintains security while preserving debugging capability Tests: 214 passing, all security tests verified
227 lines
9.2 KiB
Python
227 lines
9.2 KiB
Python
"""
|
|
Integration tests for chat summary operations.
|
|
"""
|
|
import pytest
|
|
import aiosqlite
|
|
import os
|
|
from unittest.mock import patch, AsyncMock, MagicMock
|
|
|
|
|
|
class TestChatSummaryDates:
|
|
"""Test chat summary dates endpoint."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_dates_empty(self, test_app, mock_config, admin_headers, tmp_path):
|
|
"""Test listing dates when database is empty."""
|
|
db_path = tmp_path / "test_chat.db"
|
|
|
|
# Create empty database with schema
|
|
async with aiosqlite.connect(str(db_path)) as db:
|
|
await db.execute("CREATE TABLE summaries (date TEXT, content TEXT, created_at TEXT)")
|
|
await db.commit()
|
|
|
|
with patch("routers.chat_summary.CHAT_SUMMARY_DB", str(db_path)):
|
|
response = test_app.get("/api/chat-summary/dates", headers=admin_headers)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert isinstance(data, list)
|
|
assert len(data) == 0
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_dates_with_data(self, test_app, mock_config, admin_headers, tmp_path):
|
|
"""Test listing dates with summaries."""
|
|
db_path = tmp_path / "test_chat.db"
|
|
|
|
# Create database with test data
|
|
async with aiosqlite.connect(str(db_path)) as db:
|
|
await db.execute("CREATE TABLE summaries (date TEXT, content TEXT, created_at TEXT)")
|
|
await db.execute(
|
|
"INSERT INTO summaries VALUES (?, ?, ?)", ("2024-01-01", "Summary 1", "2024-01-01T10:00:00")
|
|
)
|
|
await db.execute(
|
|
"INSERT INTO summaries VALUES (?, ?, ?)", ("2024-01-02", "Summary 2", "2024-01-02T10:00:00")
|
|
)
|
|
await db.commit()
|
|
|
|
with patch("routers.chat_summary.CHAT_SUMMARY_DB", str(db_path)):
|
|
response = test_app.get("/api/chat-summary/dates", headers=admin_headers)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert len(data) == 2
|
|
assert "2024-01-02" in data
|
|
assert "2024-01-01" in data
|
|
|
|
def test_list_dates_without_auth(self, test_app, mock_config):
|
|
"""Test listing dates without authentication."""
|
|
response = test_app.get("/api/chat-summary/dates")
|
|
|
|
assert response.status_code == 401
|
|
|
|
|
|
class TestChatSummaryGet:
|
|
"""Test get summary endpoint."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_summary_success(self, test_app, mock_config, admin_headers, tmp_path):
|
|
"""Test getting a summary for a specific date."""
|
|
db_path = tmp_path / "test_chat.db"
|
|
|
|
# Create database with test data
|
|
async with aiosqlite.connect(str(db_path)) as db:
|
|
await db.execute("CREATE TABLE summaries (date TEXT, content TEXT, created_at TEXT)")
|
|
await db.execute(
|
|
"INSERT INTO summaries VALUES (?, ?, ?)",
|
|
("2024-01-01", "Test summary content", "2024-01-01T10:00:00"),
|
|
)
|
|
await db.commit()
|
|
|
|
with patch("routers.chat_summary.CHAT_SUMMARY_DB", str(db_path)):
|
|
response = test_app.get("/api/chat-summary/summary/2024-01-01", headers=admin_headers)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["date"] == "2024-01-01"
|
|
assert data["content"] == "Test summary content"
|
|
assert data["created_at"] == "2024-01-01T10:00:00"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_summary_not_found(self, test_app, mock_config, admin_headers, tmp_path):
|
|
"""Test getting a summary for a date that doesn't exist."""
|
|
db_path = tmp_path / "test_chat.db"
|
|
|
|
# Create empty database
|
|
async with aiosqlite.connect(str(db_path)) as db:
|
|
await db.execute("CREATE TABLE summaries (date TEXT, content TEXT, created_at TEXT)")
|
|
await db.commit()
|
|
|
|
with patch("routers.chat_summary.CHAT_SUMMARY_DB", str(db_path)):
|
|
response = test_app.get("/api/chat-summary/summary/2024-01-01", headers=admin_headers)
|
|
|
|
assert response.status_code == 404
|
|
data = response.json()
|
|
assert "No summary for this date" in data["detail"]
|
|
|
|
def test_get_summary_without_auth(self, test_app, mock_config):
|
|
"""Test getting summary without authentication."""
|
|
response = test_app.get("/api/chat-summary/summary/2024-01-01")
|
|
|
|
assert response.status_code == 401
|
|
|
|
|
|
class TestChatSummaryMessages:
|
|
"""Test get messages endpoint."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_messages_success(self, test_app, mock_config, admin_headers, tmp_path):
|
|
"""Test getting messages for a specific date."""
|
|
db_path = tmp_path / "test_chat.db"
|
|
|
|
# Create database with test data
|
|
async with aiosqlite.connect(str(db_path)) as db:
|
|
await db.execute("CREATE TABLE messages (group_name TEXT, sender_name TEXT, text TEXT, timestamp TEXT)")
|
|
await db.execute(
|
|
"INSERT INTO messages VALUES (?, ?, ?, ?)",
|
|
("Group1", "User1", "Hello", "2024-01-01 10:00:00"),
|
|
)
|
|
await db.execute(
|
|
"INSERT INTO messages VALUES (?, ?, ?, ?)",
|
|
("Group1", "User2", "Hi there", "2024-01-01 10:01:00"),
|
|
)
|
|
await db.commit()
|
|
|
|
with patch("routers.chat_summary.CHAT_SUMMARY_DB", str(db_path)):
|
|
response = test_app.get("/api/chat-summary/messages/2024-01-01", headers=admin_headers)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert len(data) == 2
|
|
assert data[0]["group"] == "Group1"
|
|
assert data[0]["sender"] == "User1"
|
|
assert data[0]["text"] == "Hello"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_messages_empty(self, test_app, mock_config, admin_headers, tmp_path):
|
|
"""Test getting messages when none exist for date."""
|
|
db_path = tmp_path / "test_chat.db"
|
|
|
|
# Create empty database
|
|
async with aiosqlite.connect(str(db_path)) as db:
|
|
await db.execute("CREATE TABLE messages (group_name TEXT, sender_name TEXT, text TEXT, timestamp TEXT)")
|
|
await db.commit()
|
|
|
|
with patch("routers.chat_summary.CHAT_SUMMARY_DB", str(db_path)):
|
|
response = test_app.get("/api/chat-summary/messages/2024-01-01", headers=admin_headers)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert isinstance(data, list)
|
|
assert len(data) == 0
|
|
|
|
def test_get_messages_without_auth(self, test_app, mock_config):
|
|
"""Test getting messages without authentication."""
|
|
response = test_app.get("/api/chat-summary/messages/2024-01-01")
|
|
|
|
assert response.status_code == 401
|
|
|
|
|
|
class TestChatSummaryTrigger:
|
|
"""Test trigger summary endpoint."""
|
|
|
|
def test_trigger_summary_success(self, test_app, mock_config, admin_headers, tmp_path):
|
|
"""Test triggering a summary generation as admin."""
|
|
trigger_path = tmp_path / "trigger"
|
|
|
|
with patch.dict(os.environ, {"CHAT_SUMMARY_TRIGGER": str(trigger_path)}):
|
|
response = test_app.post("/api/chat-summary/trigger", headers=admin_headers)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["status"] == "triggered"
|
|
assert trigger_path.exists()
|
|
assert trigger_path.read_text() == "1"
|
|
|
|
def test_trigger_summary_creates_directory(self, test_app, mock_config, admin_headers, tmp_path):
|
|
"""Test that trigger creates parent directory if needed."""
|
|
trigger_path = tmp_path / "subdir" / "trigger"
|
|
|
|
with patch.dict(os.environ, {"CHAT_SUMMARY_TRIGGER": str(trigger_path)}):
|
|
response = test_app.post("/api/chat-summary/trigger", headers=admin_headers)
|
|
|
|
assert response.status_code == 200
|
|
assert trigger_path.parent.exists()
|
|
assert trigger_path.exists()
|
|
|
|
def test_trigger_summary_as_member(self, test_app, mock_config, temp_auth_file, temp_rbac_file, tmp_path):
|
|
"""Test that non-admin cannot trigger summary."""
|
|
from datetime import timedelta
|
|
|
|
from auth_service import create_access_token
|
|
|
|
member_token = create_access_token(data={"sub": "memberuser", "role": "member"}, expires_delta=timedelta(minutes=30))
|
|
headers = {"Authorization": f"Bearer {member_token}"}
|
|
|
|
trigger_path = tmp_path / "trigger"
|
|
with patch.dict(os.environ, {"CHAT_SUMMARY_TRIGGER": str(trigger_path)}):
|
|
response = test_app.post("/api/chat-summary/trigger", headers=headers)
|
|
|
|
# Page-level RBAC blocks access before endpoint-level check
|
|
assert response.status_code == 403
|
|
assert "denied" in response.json()["detail"].lower()
|
|
|
|
def test_trigger_summary_invalid_path(self, test_app, mock_config, admin_headers):
|
|
"""Test that invalid trigger paths are rejected."""
|
|
# Try to write outside /app/data/
|
|
with patch.dict(os.environ, {"CHAT_SUMMARY_TRIGGER": "/etc/passwd"}):
|
|
response = test_app.post("/api/chat-summary/trigger", headers=admin_headers)
|
|
|
|
assert response.status_code == 400
|
|
assert "Invalid trigger path" in response.json()["detail"]
|
|
|
|
def test_trigger_summary_without_auth(self, test_app, mock_config):
|
|
"""Test triggering summary without authentication."""
|
|
response = test_app.post("/api/chat-summary/trigger")
|
|
|
|
assert response.status_code == 401
|