244 lines
7.5 KiB
Python
244 lines
7.5 KiB
Python
from __future__ import annotations
|
|
|
|
from contextlib import asynccontextmanager
|
|
|
|
import structlog
|
|
from fastapi import Depends, FastAPI, File, Form, HTTPException, Request, Response, UploadFile
|
|
from fastapi.responses import FileResponse, JSONResponse
|
|
from fastapi.staticfiles import StaticFiles
|
|
|
|
from backend import auth, database, notes
|
|
from backend.config import settings
|
|
from backend.models import LoginRequest, NoteCreate, NoteUpdate, UserSelect
|
|
|
|
logger = structlog.get_logger()
|
|
|
|
|
|
@asynccontextmanager
|
|
async def lifespan(app: FastAPI):
|
|
await database.init_db()
|
|
yield
|
|
|
|
|
|
app = FastAPI(title="snotes", lifespan=lifespan)
|
|
|
|
|
|
def client_ip(request: Request) -> str | None:
|
|
fwd = request.headers.get("x-forwarded-for")
|
|
if fwd:
|
|
return fwd.split(",")[0].strip()
|
|
return request.client.host if request.client else None
|
|
|
|
|
|
def client_ua(request: Request) -> str | None:
|
|
return request.headers.get("user-agent")
|
|
|
|
|
|
def require_access(request: Request) -> None:
|
|
ip = client_ip(request)
|
|
if auth.is_lan(ip):
|
|
return
|
|
token = request.cookies.get("snotes_auth")
|
|
if token and auth.check_auth_cookie(token):
|
|
return
|
|
raise HTTPException(status_code=401, detail="authentication required")
|
|
|
|
|
|
def current_user(request: Request) -> str:
|
|
username = request.cookies.get("snotes_user")
|
|
return username or settings.user_list[0]
|
|
|
|
|
|
def valid_username(username: str) -> str:
|
|
if username not in settings.user_list and username != settings.common_space:
|
|
raise HTTPException(status_code=404, detail="unknown user")
|
|
return username
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# SPA + static
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@app.get("/", include_in_schema=False)
|
|
async def index() -> FileResponse:
|
|
return FileResponse("static/index.html")
|
|
|
|
|
|
@app.get("/favicon.ico", include_in_schema=False)
|
|
async def favicon() -> FileResponse:
|
|
return FileResponse("static/snotes-new.png")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Auth / users
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@app.get("/api/me")
|
|
async def me(request: Request) -> dict:
|
|
ip = client_ip(request)
|
|
return {
|
|
"user": current_user(request),
|
|
"common_space": settings.common_space,
|
|
"local": auth.is_lan(ip),
|
|
"authed": not auth.is_lan(ip)
|
|
and bool(
|
|
request.cookies.get("snotes_auth")
|
|
and auth.check_auth_cookie(request.cookies["snotes_auth"])
|
|
),
|
|
}
|
|
|
|
|
|
@app.get("/api/users")
|
|
async def users(_: None = Depends(require_access)) -> list[dict]:
|
|
return await database.list_users()
|
|
|
|
|
|
@app.post("/api/users/select")
|
|
async def select_user(body: UserSelect, _: None = Depends(require_access)) -> Response:
|
|
if body.username not in settings.user_list and body.username != settings.common_space:
|
|
raise HTTPException(status_code=404, detail="unknown user")
|
|
resp = JSONResponse({"username": body.username})
|
|
resp.set_cookie("snotes_user", body.username, httponly=True, samesite="lax")
|
|
return resp
|
|
|
|
|
|
@app.post("/api/auth/login")
|
|
async def login(body: LoginRequest) -> Response:
|
|
if not auth.remote_password_ok(body.password):
|
|
raise HTTPException(status_code=401, detail="invalid password")
|
|
resp = JSONResponse({"ok": True})
|
|
resp.set_cookie("snotes_auth", auth.issue_auth_cookie(), httponly=True, samesite="lax")
|
|
return resp
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Notes
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@app.post("/api/notes")
|
|
async def create_note(body: NoteCreate, request: Request, _: None = Depends(require_access)):
|
|
username = current_user(request)
|
|
if body.username:
|
|
username = valid_username(body.username)
|
|
if not body.body.strip():
|
|
raise HTTPException(status_code=400, detail="empty note")
|
|
try:
|
|
note = await notes.create_note(
|
|
username, body.body, ip=client_ip(request), ua=client_ua(request)
|
|
)
|
|
except ValueError as exc:
|
|
raise HTTPException(status_code=404, detail=str(exc)) from exc
|
|
return note
|
|
|
|
|
|
@app.get("/api/notes")
|
|
async def list_notes(request: Request, _: None = Depends(require_access)):
|
|
username = current_user(request)
|
|
return await notes.list_notes(username)
|
|
|
|
|
|
@app.get("/api/notes/recent")
|
|
async def recent(request: Request, _: None = Depends(require_access)):
|
|
return await notes.list_recent(current_user(request))
|
|
|
|
|
|
@app.get("/api/notes/{note_id}")
|
|
async def get_note(note_id: str, _: None = Depends(require_access)):
|
|
note = await notes.get_note(note_id)
|
|
if note is None:
|
|
raise HTTPException(status_code=404, detail="note not found")
|
|
return note
|
|
|
|
|
|
@app.put("/api/notes/{note_id}")
|
|
async def update_note(
|
|
note_id: str, body: NoteUpdate, request: Request, _: None = Depends(require_access)
|
|
):
|
|
note = await notes.update_note(
|
|
note_id,
|
|
body.body,
|
|
ip=client_ip(request),
|
|
ua=client_ua(request),
|
|
username=valid_username(body.username) if body.username else None,
|
|
)
|
|
if note is None:
|
|
raise HTTPException(status_code=404, detail="note not found")
|
|
return note
|
|
|
|
|
|
@app.delete("/api/notes/{note_id}")
|
|
async def delete_note(note_id: str, request: Request, _: None = Depends(require_access)):
|
|
ok = await notes.delete_note(note_id, ip=client_ip(request), ua=client_ua(request))
|
|
if not ok:
|
|
raise HTTPException(status_code=404, detail="note not found")
|
|
return {"ok": True}
|
|
|
|
|
|
@app.get("/api/notes/{note_id}/history")
|
|
async def note_history(note_id: str, _: None = Depends(require_access)):
|
|
return await notes.get_history(note_id)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Attachments
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@app.post("/api/attachments")
|
|
async def upload_attachment(
|
|
request: Request,
|
|
file: UploadFile = File(...),
|
|
username: str | None = Form(None),
|
|
_: None = Depends(require_access),
|
|
):
|
|
owner = current_user(request)
|
|
if username:
|
|
owner = valid_username(username)
|
|
data = await file.read()
|
|
if not data:
|
|
raise HTTPException(status_code=400, detail="empty file")
|
|
filename = await notes.store_attachment(owner, file.filename or "file", data)
|
|
return {
|
|
"url": f"/api/attachments/{owner}/{filename}",
|
|
"name": file.filename or filename,
|
|
"path": f"{owner}/attachments/{filename}",
|
|
}
|
|
|
|
|
|
@app.get("/api/attachments/{username}/{filename}")
|
|
async def get_attachment(username: str, filename: str, _: None = Depends(require_access)):
|
|
from pathlib import Path
|
|
|
|
full = Path(settings.data_dir) / username / "attachments" / filename
|
|
if not full.is_file():
|
|
raise HTTPException(status_code=404, detail="not found")
|
|
return FileResponse(full)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tags / search
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@app.get("/api/tags")
|
|
async def tags(_: None = Depends(require_access)):
|
|
return await notes.tag_counts()
|
|
|
|
|
|
@app.get("/api/search")
|
|
async def search(q: str, _: None = Depends(require_access)):
|
|
if not q.strip():
|
|
return []
|
|
return await notes.search_notes(q)
|
|
|
|
|
|
@app.get("/api/homepage")
|
|
async def homepage_stats(_: None = Depends(require_access)):
|
|
return await notes.get_homepage_stats()
|
|
|
|
|
|
# Mount static assets last so /api routes take precedence
|
|
app.mount("/static", StaticFiles(directory="static"), name="static")
|