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
62 lines
2.1 KiB
Python
62 lines
2.1 KiB
Python
import os
|
|
|
|
import aiosqlite
|
|
from fastapi import APIRouter, Depends, HTTPException
|
|
|
|
import auth_service as auth
|
|
from config import CHAT_SUMMARY_DB
|
|
|
|
router = APIRouter()
|
|
|
|
|
|
async def _db():
|
|
return aiosqlite.connect(CHAT_SUMMARY_DB)
|
|
|
|
|
|
@router.get("/dates")
|
|
async def list_dates():
|
|
async with await _db() as db:
|
|
cursor = await db.execute("SELECT date FROM summaries ORDER BY date DESC")
|
|
return [r[0] for r in await cursor.fetchall()]
|
|
|
|
|
|
@router.get("/summary/{date}")
|
|
async def get_summary(date: str):
|
|
async with await _db() as db:
|
|
cursor = await db.execute("SELECT content, created_at FROM summaries WHERE date=?", (date,))
|
|
row = await cursor.fetchone()
|
|
if not row:
|
|
raise HTTPException(404, "No summary for this date")
|
|
return {"date": date, "content": row[0], "created_at": row[1]}
|
|
|
|
|
|
@router.get("/messages/{date}")
|
|
async def get_messages(date: str):
|
|
async with await _db() as db:
|
|
cursor = await db.execute(
|
|
"SELECT group_name, sender_name, text, timestamp FROM messages WHERE date(timestamp)=? ORDER BY timestamp",
|
|
(date,),
|
|
)
|
|
return [{"group": r[0], "sender": r[1], "text": r[2], "time": r[3]} for r in await cursor.fetchall()]
|
|
|
|
|
|
@router.post("/trigger")
|
|
async def trigger_summary(current_user=Depends(auth.get_current_user)):
|
|
"""Trigger chat summary generation. Admin only."""
|
|
if current_user.role != "admin":
|
|
raise HTTPException(status_code=403, detail="Admin access required")
|
|
|
|
trigger_path = os.environ.get("CHAT_SUMMARY_TRIGGER", "/app/data/chat-summarizer/trigger")
|
|
|
|
# Validate path to prevent directory traversal to sensitive locations
|
|
trigger_path = os.path.abspath(trigger_path)
|
|
# Block access to system directories
|
|
forbidden_prefixes = ("/etc/", "/root/", "/sys/", "/proc/", "/boot/")
|
|
if any(trigger_path.startswith(prefix) for prefix in forbidden_prefixes):
|
|
raise HTTPException(status_code=400, detail="Invalid trigger path")
|
|
|
|
os.makedirs(os.path.dirname(trigger_path), exist_ok=True)
|
|
with open(trigger_path, "w") as f:
|
|
f.write("1")
|
|
return {"status": "triggered"}
|