Jonathan KirkwoodandClaude Opus 4.8
69f12b0519
Address the security-auditor findings the user selected (items 2, 3, 4): Default admin credentials (P2): remove the fixed `Ten31` default. First boot now generates a strong random admin password (secrets.token_urlsafe), records it 0600 at /data/.admin-password, and surfaces it once via a new "Show Initial Admin Password" StartOS action (CLI `show-admin-password`). The stored password is cleared when the admin is reset (CLI reset-password) or self-changes it (change-password endpoint). Login hardening (P2): add a per-IP in-memory sliding-window rate limiter (10 failures / 5 min -> 429 + Retry-After) in ratelimit.py; run a dummy argon2 verify when the user is unknown so timing can't enumerate usernames; keep a single generic 401 for unknown-user and wrong-password. Hardening (P3): server process now runs unprivileged -- Dockerfile adds uid 10001 appuser; start.sh (still root) chowns the mounted /data then drops via `setpriv` before exec'ing uvicorn. Spreadsheet imports are size-capped via storage.read_capped (413 past MAX_UPLOAD_SIZE) in the schedule, capital preview, and batch paths. batch_import no longer returns raw exception text (generic per-file messages). Verified in the packed amd64 container: PID1 uvicorn runs as uid 10001, /data owned 10001 with 0600 secrets; generated admin password retrievable via CLI and logs in (200); 11th bad login -> 429; admin reset clears the stored password. Tests: test_auth_hardening.py (4). Full suite 21 passed; frontend tsc + StartOS bundle clean. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
80 lines
2.5 KiB
Python
80 lines
2.5 KiB
Python
"""File storage on the data volume for uploaded documents."""
|
|
|
|
import os
|
|
import shutil
|
|
from pathlib import Path
|
|
from uuid import uuid4
|
|
|
|
from fastapi import UploadFile
|
|
|
|
from ten31portal.config import DOCS_DIR, MAX_UPLOAD_SIZE
|
|
|
|
|
|
class UploadTooLarge(Exception):
|
|
"""Raised when an upload exceeds MAX_UPLOAD_SIZE. The partial file is removed first."""
|
|
|
|
|
|
def read_capped(file: UploadFile, limit: int = MAX_UPLOAD_SIZE) -> bytes:
|
|
"""Read an upload fully into memory, but abort past `limit` instead of reading unbounded.
|
|
|
|
The spreadsheet importers must parse the whole workbook in memory; without a cap an
|
|
authenticated writer could POST a huge file and exhaust the process. Reads in 1 MB chunks
|
|
and raises UploadTooLarge (→ 413) once the total would exceed the limit.
|
|
"""
|
|
chunks: list[bytes] = []
|
|
size = 0
|
|
while chunk := file.file.read(1024 * 1024):
|
|
size += len(chunk)
|
|
if size > limit:
|
|
raise UploadTooLarge(f"Upload exceeds the {limit}-byte limit.")
|
|
chunks.append(chunk)
|
|
return b"".join(chunks)
|
|
|
|
|
|
def ensure_docs_dir() -> Path:
|
|
path = Path(DOCS_DIR)
|
|
path.mkdir(parents=True, exist_ok=True)
|
|
return path
|
|
|
|
|
|
def save_upload(file: UploadFile) -> tuple[str, int]:
|
|
"""Stream an upload to disk under an opaque name. Returns (storage_path, size_bytes).
|
|
|
|
Enforces MAX_UPLOAD_SIZE as it streams so a runaway upload can't fill the data volume;
|
|
the partial file is deleted before UploadTooLarge propagates.
|
|
"""
|
|
docs = ensure_docs_dir()
|
|
suffix = Path(file.filename or "").suffix
|
|
storage_name = f"{uuid4().hex}{suffix}"
|
|
dest = docs / storage_name
|
|
size = 0
|
|
try:
|
|
with dest.open("wb") as out:
|
|
while chunk := file.file.read(1024 * 1024):
|
|
size += len(chunk)
|
|
if size > MAX_UPLOAD_SIZE:
|
|
raise UploadTooLarge(
|
|
f"Upload exceeds the {MAX_UPLOAD_SIZE}-byte limit."
|
|
)
|
|
out.write(chunk)
|
|
except UploadTooLarge:
|
|
dest.unlink(missing_ok=True)
|
|
raise
|
|
return storage_name, size
|
|
|
|
|
|
def full_path(storage_path: str) -> Path:
|
|
"""Resolve a stored file, guarding against path traversal."""
|
|
docs = ensure_docs_dir().resolve()
|
|
candidate = (docs / storage_path).resolve()
|
|
if not str(candidate).startswith(str(docs) + os.sep):
|
|
raise ValueError("Invalid storage path")
|
|
return candidate
|
|
|
|
|
|
def delete_file(storage_path: str) -> None:
|
|
try:
|
|
full_path(storage_path).unlink(missing_ok=True)
|
|
except ValueError:
|
|
pass
|