116 lines
4.4 KiB
Python
116 lines
4.4 KiB
Python
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
|
|
from fastapi import Depends, FastAPI, Header, HTTPException, Query, Request, status
|
|
from fastapi.responses import HTMLResponse
|
|
|
|
from backend.config import Settings
|
|
from backend.db import Database
|
|
from backend.schemas import ServerConfigCreate, UserCreate
|
|
from backend.security import SecretBox, parse_user_agent
|
|
|
|
settings = Settings.from_env()
|
|
database = Database(settings.database_path, SecretBox(settings.secret_key))
|
|
app = FastAPI(title="VPN App backend", docs_url=None, redoc_url=None)
|
|
ADMIN_HTML = (Path(__file__).parent / "admin.html").read_text(encoding="utf-8")
|
|
|
|
|
|
def require_admin(authorization: str | None = Header(default=None)) -> None:
|
|
expected = f"Bearer {settings.admin_token}"
|
|
if authorization != expected:
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="invalid admin token")
|
|
|
|
|
|
@app.get("/admin", response_class=HTMLResponse)
|
|
def admin_page() -> str:
|
|
return ADMIN_HTML
|
|
|
|
|
|
@app.get("/admin/api/users", dependencies=[Depends(require_admin)])
|
|
def list_users() -> list[dict]:
|
|
return database.list_users()
|
|
|
|
|
|
@app.post("/admin/api/users", dependencies=[Depends(require_admin)])
|
|
def create_user(payload: UserCreate) -> dict:
|
|
return database.create_user(payload.model_dump())
|
|
|
|
|
|
@app.delete("/admin/api/users/{user_id}", dependencies=[Depends(require_admin)])
|
|
def delete_user(user_id: str) -> dict:
|
|
if not database.delete_user(user_id):
|
|
raise HTTPException(status_code=404, detail="user not found")
|
|
return {"deleted": user_id}
|
|
|
|
|
|
@app.post("/admin/api/users/{user_id}/devices", dependencies=[Depends(require_admin)])
|
|
def create_device(user_id: str) -> dict:
|
|
try:
|
|
return database.create_device(user_id, settings.base_url)
|
|
except KeyError as exc:
|
|
raise HTTPException(status_code=404, detail=str(exc)) from exc
|
|
|
|
|
|
@app.delete("/admin/api/devices/{device_id}", dependencies=[Depends(require_admin)])
|
|
def delete_device(device_id: str) -> dict:
|
|
if not database.delete_device(device_id):
|
|
raise HTTPException(status_code=404, detail="device not found")
|
|
return {"deleted": device_id}
|
|
|
|
|
|
@app.post("/admin/api/users/{user_id}/servers", dependencies=[Depends(require_admin)])
|
|
def upsert_server(user_id: str, payload: ServerConfigCreate) -> dict:
|
|
with database.connect() as db:
|
|
if db.execute("SELECT 1 FROM users WHERE id = ?", (user_id,)).fetchone() is None:
|
|
raise HTTPException(status_code=404, detail="user not found")
|
|
db.execute(
|
|
"INSERT INTO server_configs (id,user_id,server_name,amnezia_config) VALUES (?,?,?,?) "
|
|
"ON CONFLICT(user_id,server_name) DO UPDATE SET amnezia_config=excluded.amnezia_config",
|
|
(
|
|
user_id + ":" + payload.server_name,
|
|
user_id,
|
|
payload.server_name,
|
|
database.box.seal(payload.amnezia_config),
|
|
),
|
|
)
|
|
return {"user_id": user_id, "server_name": payload.server_name}
|
|
|
|
|
|
@app.delete(
|
|
"/admin/api/users/{user_id}/servers/{server_name}",
|
|
dependencies=[Depends(require_admin)],
|
|
)
|
|
def delete_server(user_id: str, server_name: str) -> dict:
|
|
with database.connect() as db:
|
|
cursor = db.execute(
|
|
"DELETE FROM server_configs WHERE user_id = ? AND server_name = ?",
|
|
(user_id, server_name),
|
|
)
|
|
if cursor.rowcount != 1:
|
|
raise HTTPException(status_code=404, detail="server config not found")
|
|
return {"deleted": server_name}
|
|
|
|
|
|
@app.get("/d/{device_id}")
|
|
def device_subscription(
|
|
device_id: str,
|
|
request: Request,
|
|
enroll: str | None = Query(default=None),
|
|
x_device_token: str | None = Header(default=None),
|
|
) -> dict:
|
|
token = x_device_token or enroll
|
|
if not token:
|
|
raise HTTPException(status_code=401, detail="X-Device-Token is required")
|
|
user_agent = request.headers.get("user-agent", "")
|
|
user_agent_device = parse_user_agent(user_agent)
|
|
if user_agent_device is None:
|
|
raise HTTPException(status_code=400, detail="invalid device user-agent")
|
|
try:
|
|
return database.bind_and_get_subscription(device_id, token, user_agent, user_agent_device)
|
|
except KeyError as exc:
|
|
raise HTTPException(status_code=404, detail=str(exc)) from exc
|
|
except PermissionError as exc:
|
|
raise HTTPException(status_code=403, detail=str(exc)) from exc
|
|
except ValueError as exc:
|
|
raise HTTPException(status_code=409, detail=str(exc)) from exc
|