From c92bcb0074598b08cc7a69f992a5e342b6ffa512 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E8=82=89=E6=9C=AB?= Date: Tue, 14 Jul 2026 12:54:20 +0800 Subject: [PATCH] feat: admin login, delete items, and IP bans API Co-authored-by: Cursor --- backend/app/api/admin.py | 101 ++++++++++++++++++++++++++++++++++++ backend/app/main.py | 2 + backend/app/security.py | 39 ++++++++++++++ backend/tests/test_admin.py | 32 ++++++++++++ 4 files changed, 174 insertions(+) create mode 100644 backend/app/api/admin.py create mode 100644 backend/app/security.py create mode 100644 backend/tests/test_admin.py diff --git a/backend/app/api/admin.py b/backend/app/api/admin.py new file mode 100644 index 0000000..a9fcd6c --- /dev/null +++ b/backend/app/api/admin.py @@ -0,0 +1,101 @@ +from fastapi import APIRouter, Depends, HTTPException +from pydantic import BaseModel, Field +from sqlalchemy import select +from sqlalchemy.orm import Session + +from app import config +from app.db import get_db +from app.models import BannedIP, Item +from app.security import create_access_token, require_admin +from app.services import items as items_service + +router = APIRouter(prefix="/api/admin", tags=["admin"]) + + +class LoginBody(BaseModel): + password: str + + +class BanCreate(BaseModel): + ip: str = Field(min_length=1) + reason: str = "" + + +@router.post("/login") +def login(body: LoginBody): + if body.password != config.settings.admin_password: + raise HTTPException(status_code=401, detail="Unauthorized") + return {"token": create_access_token()} + + +@router.get("/items") +def list_items( + _admin: dict = Depends(require_admin), + db: Session = Depends(get_db), +): + rows = list(db.scalars(select(Item).order_by(Item.created_at.desc())).all()) + items = [] + for item in rows: + data = items_service.item_to_dict(item) + data["id"] = item.id + data["created_ip"] = item.created_ip + data["burned"] = item.burned + items.append(data) + return {"items": items} + + +@router.delete("/items/{item_id}") +def delete_item( + item_id: int, + _admin: dict = Depends(require_admin), + db: Session = Depends(get_db), +): + item = db.get(Item, item_id) + if item is None: + raise HTTPException(status_code=404, detail="Not found") + items_service.delete_item_file(item) + db.delete(item) + db.commit() + return {"ok": True} + + +@router.get("/bans") +def list_bans( + _admin: dict = Depends(require_admin), + db: Session = Depends(get_db), +): + rows = list(db.scalars(select(BannedIP).order_by(BannedIP.created_at.desc())).all()) + return { + "bans": [ + {"ip": b.ip, "reason": b.reason, "created_at": b.created_at} for b in rows + ] + } + + +@router.post("/bans") +def create_ban( + body: BanCreate, + _admin: dict = Depends(require_admin), + db: Session = Depends(get_db), +): + existing = db.get(BannedIP, body.ip) + if existing is None: + db.add(BannedIP(ip=body.ip, reason=body.reason or "")) + else: + existing.reason = body.reason or existing.reason + db.commit() + return {"ok": True} + + +@router.delete("/bans/{ip}") +def delete_ban( + ip: str, + _admin: dict = Depends(require_admin), + db: Session = Depends(get_db), +): + ban = db.get(BannedIP, ip) + if ban is None: + raise HTTPException(status_code=404, detail="Not found") + db.delete(ban) + db.commit() + return {"ok": True} diff --git a/backend/app/main.py b/backend/app/main.py index 34f7cc3..dcbcd05 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -3,6 +3,7 @@ from contextlib import asynccontextmanager from fastapi import FastAPI +from app.api.admin import router as admin_router from app.api.public import router as public_router from app.config import settings from app.db import Base, SessionLocal, engine @@ -40,6 +41,7 @@ async def lifespan(_app: FastAPI): app = FastAPI(title="MincedPad", lifespan=lifespan) app.include_router(public_router) +app.include_router(admin_router) @app.get("/api/health") diff --git a/backend/app/security.py b/backend/app/security.py new file mode 100644 index 0000000..a47f010 --- /dev/null +++ b/backend/app/security.py @@ -0,0 +1,39 @@ +"""Admin JWT helpers (HS256).""" + +from __future__ import annotations + +from datetime import datetime, timedelta, timezone + +import jwt +from fastapi import Depends, HTTPException +from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer + +from app import config + +_bearer = HTTPBearer(auto_error=False) + + +def create_access_token(*, expires_hours: int | None = None) -> str: + hours = expires_hours if expires_hours is not None else config.settings.jwt_expire_hours + now = datetime.now(timezone.utc) + payload = { + "sub": "admin", + "iat": now, + "exp": now + timedelta(hours=hours), + } + return jwt.encode(payload, config.settings.secret_key, algorithm="HS256") + + +def verify_token(token: str) -> dict: + try: + return jwt.decode(token, config.settings.secret_key, algorithms=["HS256"]) + except jwt.PyJWTError as exc: + raise HTTPException(status_code=401, detail="Unauthorized") from exc + + +def require_admin( + creds: HTTPAuthorizationCredentials | None = Depends(_bearer), +) -> dict: + if creds is None or creds.scheme.lower() != "bearer" or not creds.credentials: + raise HTTPException(status_code=401, detail="Unauthorized") + return verify_token(creds.credentials) diff --git a/backend/tests/test_admin.py b/backend/tests/test_admin.py new file mode 100644 index 0000000..e97bcdb --- /dev/null +++ b/backend/tests/test_admin.py @@ -0,0 +1,32 @@ +def test_admin_login_and_delete(client): + token = client.post("/api/admin/login", json={"password": "changeme"}).json()["token"] + headers = {"Authorization": f"Bearer {token}"} + slug = client.post("/api/items", json={"body": "bye"}).json()["slug"] + items = client.get("/api/admin/items", headers=headers).json()["items"] + item_id = next(i["id"] for i in items if i["slug"] == slug) + assert client.delete(f"/api/admin/items/{item_id}", headers=headers).status_code == 200 + assert client.get(f"/api/items/{slug}").status_code == 404 + + +def test_admin_login_rejects_bad_password(client): + r = client.post("/api/admin/login", json={"password": "wrong"}) + assert r.status_code == 401 + + +def test_admin_requires_bearer(client): + assert client.get("/api/admin/items").status_code == 401 + assert client.get("/api/admin/items", headers={"Authorization": "Bearer nope"}).status_code == 401 + + +def test_admin_ban_and_unban(client): + token = client.post("/api/admin/login", json={"password": "changeme"}).json()["token"] + headers = {"Authorization": f"Bearer {token}"} + + r = client.post("/api/admin/bans", headers=headers, json={"ip": "1.2.3.4", "reason": "spam"}) + assert r.status_code == 200 + bans = client.get("/api/admin/bans", headers=headers).json()["bans"] + assert any(b["ip"] == "1.2.3.4" and b.get("reason") == "spam" for b in bans) + + assert client.delete("/api/admin/bans/1.2.3.4", headers=headers).status_code == 200 + bans_after = client.get("/api/admin/bans", headers=headers).json()["bans"] + assert all(b["ip"] != "1.2.3.4" for b in bans_after)