PRIVATE_SELF_DEV/Forensic-Mail-Indexer
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
main
Forensic-Mail-Indexer/OLD_VERSIONS/v2.0/pst-indexer/backend/app.py
T

836 lines
29 KiB
Python
Raw Permalink Normal View History

v_2.2
2026-09-13 20:07:03 +01:00
"""
PST Indexer — Main application (port 8000).
Serves:
- The web UI (login, setup wizard, archive browser)
- REST API for PST upload, indexing, search
- All data-access endpoints require an authenticated session
First-run behavior:
- If no users exist, the UI routes users to a setup wizard.
- /api/setup is only callable while the users table is empty; after the first
admin is created, the endpoint returns 409.
"""
import os
import re
import sqlite3
import uuid
import asyncio
from datetime import datetime, timezone
from pathlib import Path
from typing import Optional
import aiofiles
import pypff
from fastapi import (
BackgroundTasks, Depends, FastAPI, File, HTTPException,
Request, Response, UploadFile, status,
)
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import JSONResponse, FileResponse
from fastapi.staticfiles import StaticFiles
from pydantic import BaseModel
import auth
from auth import (
SESSION_COOKIE_NAME, SESSION_TTL_SECONDS,
User, count_users, count_admins, create_user, find_user_by_username,
find_user_by_id, get_totp_secret, verify_password, verify_totp,
record_login, create_session_token, login_limiter, require_app_user,
get_db,
)
# ---------------------------------------------------------------------------
# Config
# ---------------------------------------------------------------------------
DATA_DIR = Path(os.environ.get("DATA_DIR", "/app/data"))
UPLOAD_DIR = DATA_DIR / "uploads"
FRONTEND_DIR = Path(os.environ.get("FRONTEND_DIR", "/app/frontend"))
UPLOAD_DIR.mkdir(parents=True, exist_ok=True)
DATA_DIR.mkdir(parents=True, exist_ok=True)
# Cookie security. Flip to True when serving over HTTPS.
COOKIE_SECURE = os.environ.get("COOKIE_SECURE", "false").lower() == "true"
# ---------------------------------------------------------------------------
# App DB schema (messages + pst_files)
# ---------------------------------------------------------------------------
def init_app_schema():
with get_db() as conn:
conn.executescript("""
CREATE TABLE IF NOT EXISTS pst_folders (
id TEXT PRIMARY KEY,
name TEXT NOT NULL,
created_at TEXT NOT NULL,
created_by TEXT
);
CREATE TABLE IF NOT EXISTS folder_permissions (
user_id TEXT NOT NULL,
folder_id TEXT NOT NULL,
granted_at TEXT NOT NULL,
granted_by TEXT,
PRIMARY KEY (user_id, folder_id)
);
CREATE INDEX IF NOT EXISTS idx_folder_perms_user ON folder_permissions(user_id);
CREATE INDEX IF NOT EXISTS idx_folder_perms_folder ON folder_permissions(folder_id);
CREATE TABLE IF NOT EXISTS pst_files (
id TEXT PRIMARY KEY,
filename TEXT NOT NULL,
original_name TEXT NOT NULL,
uploaded_at TEXT NOT NULL,
size_bytes INTEGER NOT NULL,
status TEXT NOT NULL DEFAULT 'pending',
message_count INTEGER DEFAULT 0,
error TEXT,
uploaded_by TEXT,
folder_id TEXT
);
CREATE TABLE IF NOT EXISTS messages (
id INTEGER PRIMARY KEY AUTOINCREMENT,
pst_id TEXT NOT NULL,
folder TEXT,
subject TEXT,
sender TEXT,
recipients TEXT,
sent_date TEXT,
body TEXT,
attachments TEXT,
FOREIGN KEY (pst_id) REFERENCES pst_files(id) ON DELETE CASCADE
);
CREATE INDEX IF NOT EXISTS idx_messages_pst ON messages(pst_id);
CREATE INDEX IF NOT EXISTS idx_messages_sent_date ON messages(sent_date);
CREATE INDEX IF NOT EXISTS idx_pst_files_folder ON pst_files(folder_id);
CREATE VIRTUAL TABLE IF NOT EXISTS messages_fts USING fts5(
subject, sender, recipients, body, folder,
content='messages',
content_rowid='id',
tokenize='porter unicode61'
);
CREATE TRIGGER IF NOT EXISTS messages_ai AFTER INSERT ON messages BEGIN
INSERT INTO messages_fts(rowid, subject, sender, recipients, body, folder)
VALUES (new.id, new.subject, new.sender, new.recipients, new.body, new.folder);
END;
CREATE TRIGGER IF NOT EXISTS messages_ad AFTER DELETE ON messages BEGIN
INSERT INTO messages_fts(messages_fts, rowid, subject, sender, recipients, body, folder)
VALUES ('delete', old.id, old.subject, old.sender, old.recipients, old.body, old.folder);
END;
""")
# Backfill: add folder_id column to DBs created before this feature existed.
cols = [r["name"] for r in conn.execute("PRAGMA table_info(pst_files)").fetchall()]
if "folder_id" not in cols:
conn.execute("ALTER TABLE pst_files ADD COLUMN folder_id TEXT")
conn.execute("CREATE INDEX IF NOT EXISTS idx_pst_files_folder ON pst_files(folder_id)")
conn.commit()
# ---------------------------------------------------------------------------
# Access control helpers — which folders / PSTs is a given user allowed to see?
#
# Rules:
# - Admins see everything.
# - Regular users see only folders they have been explicitly granted,
# plus the implicit "Uncategorized" bucket (folder_id = NULL) — this is
# deliberately shared because otherwise newly-uploaded PSTs would be
# invisible until someone moved them into a permitted folder.
# - A PST is visible iff its folder is visible (or the PST is uncategorized).
# ---------------------------------------------------------------------------
def allowed_folder_ids_for(user) -> set[str]:
"""Return the set of folder_ids a user is explicitly permitted on.
For admins this is not used — they see everything unconditionally."""
with get_db() as conn:
rows = conn.execute(
"SELECT folder_id FROM folder_permissions WHERE user_id = ?",
(user.id,),
).fetchall()
return {r["folder_id"] for r in rows}
def user_can_see_folder(user, folder_id: Optional[str]) -> bool:
"""Can this user see content inside this folder? folder_id may be None
(Uncategorized), which is visible to everyone."""
if user.is_admin:
return True
if folder_id is None:
return True # Uncategorized is shared
return folder_id in allowed_folder_ids_for(user)
def user_can_see_pst(user, pst_id: str) -> bool:
if user.is_admin:
return True
with get_db() as conn:
row = conn.execute(
"SELECT folder_id FROM pst_files WHERE id = ?", (pst_id,)
).fetchone()
if not row:
return False
return user_can_see_folder(user, row["folder_id"])
# ---------------------------------------------------------------------------
# PST indexing helpers (unchanged from pre-auth version)
# ---------------------------------------------------------------------------
def _safe_str(value) -> str:
if value is None:
return ""
if isinstance(value, bytes):
try:
return value.decode("utf-8", errors="replace")
except Exception:
return value.decode("latin-1", errors="replace")
return str(value)
def _extract_recipients(message) -> str:
parts = []
try:
headers = _safe_str(getattr(message, "transport_headers", ""))
if headers:
for line in headers.split("\n"):
low = line.lower()
if low.startswith("to:") or low.startswith("cc:") or low.startswith("bcc:"):
parts.append(line.strip())
except Exception:
pass
return " | ".join(parts) if parts else ""
def _extract_body(message) -> str:
for attr in ("plain_text_body", "html_body", "rtf_body"):
try:
body = getattr(message, attr, None)
if body:
text = _safe_str(body)
if attr == "html_body":
text = re.sub(r"<[^>]+>", " ", text)
text = re.sub(r"\s+", " ", text).strip()
if text:
return text
except Exception:
continue
return ""
def _extract_sender(message) -> str:
for attr in ("sender_name", "sender_email_address"):
try:
v = _safe_str(getattr(message, attr, ""))
if v:
return v
except Exception:
continue
try:
headers = _safe_str(getattr(message, "transport_headers", ""))
for line in headers.split("\n"):
if line.lower().startswith("from:"):
return line.split(":", 1)[1].strip()
except Exception:
pass
return ""
def _extract_date(message) -> Optional[str]:
try:
d = getattr(message, "delivery_time", None) or getattr(message, "client_submit_time", None)
if d:
if isinstance(d, datetime):
return d.isoformat()
return _safe_str(d)
except Exception:
pass
return None
def _extract_attachments(message) -> str:
names = []
try:
count = getattr(message, "number_of_attachments", 0)
for i in range(count):
try:
att = message.get_attachment(i)
name = _safe_str(getattr(att, "name", "") or getattr(att, "long_filename", ""))
if name:
names.append(name)
except Exception:
continue
except Exception:
pass
return " | ".join(names)
def _walk_folders(folder, path=""):
try:
name = _safe_str(getattr(folder, "name", "") or "")
except Exception:
name = ""
current_path = f"{path}/{name}" if name else (path or "/")
try:
for i in range(folder.number_of_sub_messages):
try:
msg = folder.get_sub_message(i)
yield current_path, msg
except Exception:
continue
except Exception:
pass
try:
for i in range(folder.number_of_sub_folders):
try:
sub = folder.get_sub_folder(i)
yield from _walk_folders(sub, current_path)
except Exception:
continue
except Exception:
pass
def index_pst_file(pst_id: str, file_path: Path):
pff = None
try:
pff = pypff.file()
pff.open(str(file_path))
root = pff.get_root_folder()
count = 0
batch = []
BATCH_SIZE = 500
with get_db() as conn:
for folder_path, msg in _walk_folders(root):
try:
batch.append((
pst_id, folder_path,
_safe_str(getattr(msg, "subject", "")),
_extract_sender(msg),
_extract_recipients(msg),
_extract_date(msg),
_extract_body(msg),
_extract_attachments(msg),
))
count += 1
if len(batch) >= BATCH_SIZE:
conn.executemany(
"""INSERT INTO messages
(pst_id, folder, subject, sender, recipients, sent_date, body, attachments)
VALUES (?, ?, ?, ?, ?, ?, ?, ?)""",
batch,
)
conn.commit()
batch.clear()
except Exception as e:
print(f"[index] skip: {e}")
continue
if batch:
conn.executemany(
"""INSERT INTO messages
(pst_id, folder, subject, sender, recipients, sent_date, body, attachments)
VALUES (?, ?, ?, ?, ?, ?, ?, ?)""",
batch,
)
conn.commit()
conn.execute(
"UPDATE pst_files SET status = ?, message_count = ? WHERE id = ?",
("ready", count, pst_id),
)
conn.commit()
print(f"[index] {pst_id}: {count} messages")
except Exception as e:
print(f"[index] failed: {e}")
with get_db() as conn:
conn.execute(
"UPDATE pst_files SET status = ?, error = ? WHERE id = ?",
("failed", str(e), pst_id),
)
conn.commit()
finally:
if pff is not None:
try:
pff.close()
except Exception:
pass
# ---------------------------------------------------------------------------
# FastAPI
# ---------------------------------------------------------------------------
app = FastAPI(title="PST Indexer")
app.add_middleware(
CORSMiddleware,
allow_origins=[],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
@app.on_event("startup")
def _startup():
auth.init_auth_schema()
init_app_schema()
def _set_session_cookie(response: Response, token: str):
response.set_cookie(
key=SESSION_COOKIE_NAME,
value=token,
max_age=SESSION_TTL_SECONDS,
httponly=True,
secure=COOKIE_SECURE,
samesite="lax",
path="/",
)
def _clear_session_cookie(response: Response):
response.delete_cookie(key=SESSION_COOKIE_NAME, path="/")
# ----- Setup / auth endpoints -----
class SetupRequest(BaseModel):
username: str
password: str
class LoginRequest(BaseModel):
username: str
password: str
totp_code: Optional[str] = None
@app.get("/api/setup/status")
def setup_status():
"""Lets the frontend decide whether to show the setup wizard."""
return {"needs_setup": count_users() == 0}
@app.post("/api/setup")
def setup_admin(payload: SetupRequest, response: Response):
if count_users() > 0:
raise HTTPException(status.HTTP_409_CONFLICT, "Setup has already been completed")
try:
user = create_user(payload.username, payload.password, role="admin")
except ValueError as e:
raise HTTPException(400, str(e))
record_login(user.id)
token = create_session_token(user, scope="app")
_set_session_cookie(response, token)
return {"user": user.public_dict()}
@app.post("/api/login")
def login(payload: LoginRequest, request: Request, response: Response):
ip = request.client.host if request.client else "unknown"
rate_key = f"{ip}:{payload.username.lower()}"
if not login_limiter.check(rate_key):
raise HTTPException(429, "Too many login attempts. Please wait a few minutes.")
found = find_user_by_username(payload.username)
if not found:
login_limiter.record(rate_key)
raise HTTPException(401, "Invalid credentials")
user, stored_pw_hash = found
if not verify_password(stored_pw_hash, payload.password):
login_limiter.record(rate_key)
raise HTTPException(401, "Invalid credentials")
if user.mfa_enabled:
if not payload.totp_code:
# Tell the client MFA is required
return JSONResponse(
status_code=401,
content={"detail": "MFA required", "mfa_required": True},
)
secret = get_totp_secret(user.id)
if not secret or not verify_totp(secret, payload.totp_code):
login_limiter.record(rate_key)
raise HTTPException(401, "Invalid MFA code")
login_limiter.reset(rate_key)
record_login(user.id)
token = create_session_token(user, scope="app")
_set_session_cookie(response, token)
return {"user": user.public_dict()}
@app.post("/api/logout")
def logout(response: Response):
_clear_session_cookie(response)
return {"ok": True}
@app.get("/api/me")
def me(user: User = Depends(require_app_user)):
return {"user": user.public_dict()}
# ----- PST endpoints (all require auth) -----
@app.get("/api/health")
def health():
return {"status": "ok"}
@app.get("/api/pst-files")
def list_pst_files(user: User = Depends(require_app_user)):
with get_db() as conn:
if user.is_admin:
rows = conn.execute(
"""SELECT id, original_name, uploaded_at, size_bytes, status,
message_count, error, folder_id
FROM pst_files ORDER BY uploaded_at DESC"""
).fetchall()
else:
allowed = allowed_folder_ids_for(user)
if allowed:
placeholders = ",".join("?" for _ in allowed)
rows = conn.execute(
f"""SELECT id, original_name, uploaded_at, size_bytes, status,
message_count, error, folder_id
FROM pst_files
WHERE folder_id IS NULL OR folder_id IN ({placeholders})
ORDER BY uploaded_at DESC""",
list(allowed),
).fetchall()
else:
rows = conn.execute(
"""SELECT id, original_name, uploaded_at, size_bytes, status,
message_count, error, folder_id
FROM pst_files WHERE folder_id IS NULL
ORDER BY uploaded_at DESC"""
).fetchall()
return [dict(r) for r in rows]
# ----- Folder endpoints (shared with admin panel) -----
class FolderCreate(BaseModel):
name: str
class FolderRename(BaseModel):
name: str
class FolderAssign(BaseModel):
folder_id: Optional[str] = None # null = move to uncategorized
def _validate_folder_name(name: str) -> str:
name = (name or "").strip()
if not name:
raise HTTPException(400, "Folder name cannot be empty")
if len(name) > 80:
raise HTTPException(400, "Folder name too long (max 80 characters)")
return name
@app.get("/api/folders")
def list_folders(user: User = Depends(require_app_user)):
with get_db() as conn:
if user.is_admin:
rows = conn.execute(
"""SELECT f.id, f.name, f.created_at, f.created_by,
COUNT(p.id) AS file_count
FROM pst_folders f
LEFT JOIN pst_files p ON p.folder_id = f.id
GROUP BY f.id
ORDER BY f.name COLLATE NOCASE"""
).fetchall()
else:
allowed = allowed_folder_ids_for(user)
if not allowed:
return []
placeholders = ",".join("?" for _ in allowed)
rows = conn.execute(
f"""SELECT f.id, f.name, f.created_at, f.created_by,
COUNT(p.id) AS file_count
FROM pst_folders f
LEFT JOIN pst_files p ON p.folder_id = f.id
WHERE f.id IN ({placeholders})
GROUP BY f.id
ORDER BY f.name COLLATE NOCASE""",
list(allowed),
).fetchall()
return [dict(r) for r in rows]
@app.post("/api/folders")
def create_folder(payload: FolderCreate, user: User = Depends(require_app_user)):
# Only admins can create folders — the admin panel is the authoritative
# management surface. (We still allow creation from the main app sidebar,
# but only for admin users.)
if not user.is_admin:
raise HTTPException(403, "Only administrators can create folders")
name = _validate_folder_name(payload.name)
with get_db() as conn:
existing = conn.execute(
"SELECT id FROM pst_folders WHERE name = ? COLLATE NOCASE", (name,)
).fetchone()
if existing:
raise HTTPException(409, "A folder with that name already exists")
folder_id = uuid.uuid4().hex
conn.execute(
"""INSERT INTO pst_folders (id, name, created_at, created_by)
VALUES (?, ?, ?, ?)""",
(folder_id, name, datetime.now(timezone.utc).isoformat(), user.id),
)
conn.commit()
return {"id": folder_id, "name": name, "file_count": 0}
@app.put("/api/folders/{folder_id}")
def rename_folder(folder_id: str, payload: FolderRename,
user: User = Depends(require_app_user)):
if not user.is_admin:
raise HTTPException(403, "Only administrators can rename folders")
name = _validate_folder_name(payload.name)
with get_db() as conn:
if not conn.execute("SELECT id FROM pst_folders WHERE id = ?",
(folder_id,)).fetchone():
raise HTTPException(404, "Folder not found")
clash = conn.execute(
"SELECT id FROM pst_folders WHERE name = ? COLLATE NOCASE AND id != ?",
(name, folder_id),
).fetchone()
if clash:
raise HTTPException(409, "A folder with that name already exists")
conn.execute("UPDATE pst_folders SET name = ? WHERE id = ?",
(name, folder_id))
conn.commit()
return {"id": folder_id, "name": name}
@app.delete("/api/folders/{folder_id}")
def delete_folder(folder_id: str, user: User = Depends(require_app_user)):
"""Delete a folder. Any PSTs inside it become uncategorized (folder_id = NULL)."""
if not user.is_admin:
raise HTTPException(403, "Only administrators can delete folders")
with get_db() as conn:
if not conn.execute("SELECT id FROM pst_folders WHERE id = ?",
(folder_id,)).fetchone():
raise HTTPException(404, "Folder not found")
conn.execute("UPDATE pst_files SET folder_id = NULL WHERE folder_id = ?",
(folder_id,))
conn.execute("DELETE FROM folder_permissions WHERE folder_id = ?",
(folder_id,))
conn.execute("DELETE FROM pst_folders WHERE id = ?", (folder_id,))
conn.commit()
return {"deleted": folder_id}
@app.put("/api/pst-files/{pst_id}/folder")
def assign_pst_to_folder(pst_id: str, payload: FolderAssign,
user: User = Depends(require_app_user)):
# Check the user can see the PST's current location AND the destination
if not user_can_see_pst(user, pst_id):
raise HTTPException(404, "PST not found")
if payload.folder_id and not user_can_see_folder(user, payload.folder_id):
raise HTTPException(403, "You do not have access to the target folder")
with get_db() as conn:
if not conn.execute("SELECT id FROM pst_files WHERE id = ?",
(pst_id,)).fetchone():
raise HTTPException(404, "PST not found")
if payload.folder_id:
if not conn.execute("SELECT id FROM pst_folders WHERE id = ?",
(payload.folder_id,)).fetchone():
raise HTTPException(404, "Folder not found")
conn.execute("UPDATE pst_files SET folder_id = ? WHERE id = ?",
(payload.folder_id, pst_id))
conn.commit()
return {"pst_id": pst_id, "folder_id": payload.folder_id}
@app.post("/api/pst-files")
async def upload_pst(
background_tasks: BackgroundTasks,
file: UploadFile = File(...),
folder_id: Optional[str] = None,
user: User = Depends(require_app_user),
):
if not file.filename:
raise HTTPException(400, "No filename provided")
if not file.filename.lower().endswith((".pst", ".ost")):
raise HTTPException(400, "Only .pst and .ost files are supported")
# Validate folder if provided (and verify user has access to it)
if folder_id:
with get_db() as conn:
exists = conn.execute("SELECT id FROM pst_folders WHERE id = ?",
(folder_id,)).fetchone()
if not exists:
raise HTTPException(400, "Target folder does not exist")
if not user_can_see_folder(user, folder_id):
raise HTTPException(403, "You do not have access to the target folder")
pst_id = uuid.uuid4().hex
stored_name = f"{pst_id}_{file.filename}"
stored_path = UPLOAD_DIR / stored_name
size = 0
async with aiofiles.open(stored_path, "wb") as out:
while chunk := await file.read(1024 * 1024):
await out.write(chunk)
size += len(chunk)
with get_db() as conn:
conn.execute(
"""INSERT INTO pst_files
(id, filename, original_name, uploaded_at, size_bytes, status,
uploaded_by, folder_id)
VALUES (?, ?, ?, ?, ?, 'indexing', ?, ?)""",
(pst_id, stored_name, file.filename,
datetime.now(timezone.utc).isoformat(), size, user.id, folder_id),
)
conn.commit()
background_tasks.add_task(asyncio.to_thread, index_pst_file, pst_id, stored_path)
return {"id": pst_id, "original_name": file.filename,
"status": "indexing", "size_bytes": size, "folder_id": folder_id}
@app.delete("/api/pst-files/{pst_id}")
def delete_pst(pst_id: str, user: User = Depends(require_app_user)):
if not user_can_see_pst(user, pst_id):
raise HTTPException(404, "PST not found")
with get_db() as conn:
row = conn.execute(
"SELECT filename FROM pst_files WHERE id = ?", (pst_id,)
).fetchone()
if not row:
raise HTTPException(404, "PST not found")
conn.execute("DELETE FROM messages WHERE pst_id = ?", (pst_id,))
conn.execute("DELETE FROM pst_files WHERE id = ?", (pst_id,))
conn.commit()
try:
(UPLOAD_DIR / row["filename"]).unlink(missing_ok=True)
except Exception:
pass
return {"deleted": pst_id}
@app.get("/api/pst-files/{pst_id}/folders")
def get_folders(pst_id: str, user: User = Depends(require_app_user)):
if not user_can_see_pst(user, pst_id):
raise HTTPException(404, "PST not found")
with get_db() as conn:
rows = conn.execute(
"""SELECT folder, COUNT(*) as count FROM messages
WHERE pst_id = ? GROUP BY folder ORDER BY folder""",
(pst_id,),
).fetchall()
return [dict(r) for r in rows]
def _escape_fts(query: str) -> str:
tokens = []
for raw in query.split():
if re.fullmatch(r"[A-Za-z0-9]+", raw):
tokens.append(f'"{raw}"*')
else:
sanitized = raw.replace('"', '""')
tokens.append(f'"{sanitized}"')
return " ".join(tokens)
@app.get("/api/pst-files/{pst_id}/messages")
def search_messages(
pst_id: str,
q: Optional[str] = None,
folder: Optional[str] = None,
limit: int = 50,
offset: int = 0,
user: User = Depends(require_app_user),
):
if not user_can_see_pst(user, pst_id):
raise HTTPException(404, "PST not found")
limit = max(1, min(limit, 200))
offset = max(0, offset)
with get_db() as conn:
p = conn.execute("SELECT id FROM pst_files WHERE id = ?", (pst_id,)).fetchone()
if not p:
raise HTTPException(404, "PST not found")
params = [pst_id]
if q and q.strip():
fts_q = _escape_fts(q.strip())
base = """
FROM messages m
JOIN messages_fts f ON f.rowid = m.id
WHERE m.pst_id = ? AND messages_fts MATCH ?
"""
params.append(fts_q)
if folder:
base += " AND m.folder = ?"
params.append(folder)
total = conn.execute("SELECT COUNT(*) as c " + base, params).fetchone()["c"]
data_sql = (
"SELECT m.id, m.folder, m.subject, m.sender, m.recipients, m.sent_date, "
" m.attachments, snippet(messages_fts, 3, '<mark>', '</mark>', '…', 24) AS snippet "
+ base + " ORDER BY rank LIMIT ? OFFSET ?"
)
rows = conn.execute(data_sql, params + [limit, offset]).fetchall()
else:
where = ["m.pst_id = ?"]
if folder:
where.append("m.folder = ?")
params.append(folder)
where_sql = " AND ".join(where)
total = conn.execute(
f"SELECT COUNT(*) as c FROM messages m WHERE {where_sql}", params
).fetchone()["c"]
rows = conn.execute(
f"""SELECT m.id, m.folder, m.subject, m.sender, m.recipients,
m.sent_date, m.attachments, substr(m.body, 1, 200) AS snippet
FROM messages m WHERE {where_sql}
ORDER BY m.sent_date DESC LIMIT ? OFFSET ?""",
params + [limit, offset],
).fetchall()
return {"total": total, "limit": limit, "offset": offset,
"items": [dict(r) for r in rows]}
@app.get("/api/messages/{message_id}")
def get_message(message_id: int, user: User = Depends(require_app_user)):
with get_db() as conn:
row = conn.execute(
"""SELECT id, pst_id, folder, subject, sender, recipients,
sent_date, body, attachments
FROM messages WHERE id = ?""",
(message_id,),
).fetchone()
if not row:
raise HTTPException(404, "Message not found")
if not user_can_see_pst(user, row["pst_id"]):
raise HTTPException(404, "Message not found")
return dict(row)
# ---------------------------------------------------------------------------
# Static frontend (anonymous access — but index.html enforces auth client-side
# AND the APIs enforce it server-side, which is what actually matters)
# ---------------------------------------------------------------------------
if FRONTEND_DIR.exists():
app.mount("/", StaticFiles(directory=str(FRONTEND_DIR), html=True), name="frontend")
Reference in New Issue Copy Permalink