623 lines
27 KiB
Python
623 lines
27 KiB
Python
"""HTTP-сервис certgen: авторизация, API и веб-интерфейс."""
|
|
from __future__ import annotations
|
|
|
|
import base64
|
|
import json
|
|
import re
|
|
import tempfile
|
|
import threading
|
|
import time
|
|
import traceback
|
|
from collections import defaultdict, deque
|
|
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
|
from pathlib import Path
|
|
from urllib.parse import parse_qs, quote, unquote, urlparse
|
|
|
|
from . import __version__, auth, config, store, ui
|
|
from .render import render_certificate
|
|
from .service import build_values, cert_api, create_certificate
|
|
|
|
SLUG_RE = re.compile(r"^[a-zA-Z0-9_-]{1,64}$")
|
|
CERT_FILE_RE = re.compile(r"^/c/([^/]+?)\.(png|pdf)$")
|
|
CERT_PAGE_RE = re.compile(r"^/c/([^/]+)$")
|
|
|
|
COOKIE_NAME = "certgen_session"
|
|
|
|
# Простейший ограничитель попыток входа: ip -> очередь меток времени.
|
|
_LOGIN_ATTEMPTS: dict[str, deque] = defaultdict(deque)
|
|
_LOGIN_LOCK = threading.Lock()
|
|
LOGIN_MAX = 10
|
|
LOGIN_WINDOW = 300 # секунд
|
|
|
|
|
|
def _login_allowed(ip: str) -> bool:
|
|
now = time.time()
|
|
with _LOGIN_LOCK:
|
|
q = _LOGIN_ATTEMPTS[ip]
|
|
while q and now - q[0] > LOGIN_WINDOW:
|
|
q.popleft()
|
|
if len(q) >= LOGIN_MAX:
|
|
return False
|
|
q.append(now)
|
|
return True
|
|
|
|
|
|
def _base_url(handler: BaseHTTPRequestHandler) -> str:
|
|
if config.BASE_URL:
|
|
return config.BASE_URL
|
|
proto = handler.headers.get("X-Forwarded-Proto", "http")
|
|
host = handler.headers.get("Host") or f"{config.HOST}:{config.PORT}"
|
|
return f"{proto}://{host}"
|
|
|
|
|
|
def _public_cert(cert: dict) -> dict:
|
|
"""Убрать из ответа локальные пути к файлам."""
|
|
out = dict(cert)
|
|
out.pop("png_path", None)
|
|
out.pop("pdf_path", None)
|
|
return out
|
|
|
|
|
|
class Handler(BaseHTTPRequestHandler):
|
|
server_version = f"certgen/{__version__}"
|
|
|
|
# ------------------------------------------------------------ низкий уровень
|
|
def _send(self, code: int, content_type: str, body: bytes, extra: dict | None = None):
|
|
self.send_response(code)
|
|
self.send_header("Content-Type", content_type)
|
|
self.send_header("Content-Length", str(len(body)))
|
|
for k, v in (extra or {}).items():
|
|
self.send_header(k, v)
|
|
self.end_headers()
|
|
if self.command != "HEAD":
|
|
self.wfile.write(body)
|
|
|
|
def _json(self, obj, code: int = 200, extra: dict | None = None):
|
|
self._send(code, "application/json; charset=utf-8",
|
|
json.dumps(obj, ensure_ascii=False, indent=2).encode("utf-8"), extra)
|
|
|
|
def _error(self, code: int, message: str):
|
|
self._json({"error": message}, code)
|
|
|
|
def _redirect(self, location: str, extra: dict | None = None):
|
|
self._send(302, "text/plain; charset=utf-8", b"", {"Location": location, **(extra or {})})
|
|
|
|
def _html(self, markup: str, code: int = 200, extra: dict | None = None):
|
|
self._send(code, "text/html; charset=utf-8", markup.encode("utf-8"), extra)
|
|
|
|
def _read_body(self) -> dict:
|
|
length = int(self.headers.get("Content-Length") or 0)
|
|
if not length:
|
|
return {}
|
|
raw = self.rfile.read(length)
|
|
try:
|
|
data = json.loads(raw.decode("utf-8"))
|
|
return data if isinstance(data, dict) else {}
|
|
except Exception:
|
|
return {}
|
|
|
|
def log_message(self, fmt, *args):
|
|
print(f"[certgen] {self.address_string()} - {fmt % args}", flush=True)
|
|
|
|
# ------------------------------------------------------------------- сессии
|
|
def _cookies(self) -> dict:
|
|
out = {}
|
|
for part in (self.headers.get("Cookie") or "").split(";"):
|
|
if "=" in part:
|
|
k, v = part.strip().split("=", 1)
|
|
out[k] = v
|
|
return out
|
|
|
|
def _current_user(self) -> dict | None:
|
|
token = self._cookies().get(COOKIE_NAME)
|
|
return store.get_session_user(token) if token else None
|
|
|
|
def _is_https(self) -> bool:
|
|
if config.BASE_URL.startswith("https://"):
|
|
return True
|
|
return (self.headers.get("X-Forwarded-Proto", "").lower() == "https")
|
|
|
|
def _session_cookie(self, token: str) -> str:
|
|
cookie = (f"{COOKIE_NAME}={token}; Path=/; HttpOnly; SameSite=Lax; "
|
|
f"Max-Age={config.SESSION_TTL_DAYS * 86400}")
|
|
if self._is_https():
|
|
cookie += "; Secure"
|
|
return cookie
|
|
|
|
def _clear_cookie(self) -> str:
|
|
cookie = f"{COOKIE_NAME}=; Path=/; HttpOnly; SameSite=Lax; Max-Age=0"
|
|
if self._is_https():
|
|
cookie += "; Secure"
|
|
return cookie
|
|
|
|
def _start_session(self, user: dict) -> str:
|
|
token = auth.new_token()
|
|
store.create_session(token, user["id"], config.SESSION_TTL_DAYS)
|
|
return self._session_cookie(token)
|
|
|
|
# ------------------------------------------------------------------ доступ
|
|
def _require_api_user(self) -> dict | None:
|
|
user = self._current_user()
|
|
if not user:
|
|
self._error(401, "Требуется вход")
|
|
return None
|
|
return user
|
|
|
|
def _require_api_admin(self) -> dict | None:
|
|
user = self._require_api_user()
|
|
if user and user["role"] != "admin":
|
|
self._error(403, "Доступно только администратору")
|
|
return None
|
|
return user
|
|
|
|
def _can_access(self, user: dict, cert: dict) -> bool:
|
|
if user["role"] == "admin":
|
|
return True
|
|
recipient = (cert.get("recipient_email") or "").lower()
|
|
return bool(recipient) and recipient == user["email"].lower()
|
|
|
|
def _cert_accessible(self, user: dict, ref: str) -> dict | None:
|
|
cert = store.get_certificate(ref)
|
|
if cert and self._can_access(user, cert):
|
|
return cert
|
|
return None
|
|
|
|
# --------------------------------------------------------------------- GET
|
|
def do_GET(self):
|
|
try:
|
|
self._route_get()
|
|
except Exception as exc: # noqa: BLE001
|
|
traceback.print_exc()
|
|
self._error(500, str(exc))
|
|
|
|
def _route_get(self):
|
|
parsed = urlparse(self.path)
|
|
path = unquote(parsed.path)
|
|
base = _base_url(self)
|
|
user = self._current_user()
|
|
|
|
if path == "/health":
|
|
return self._json({"ok": True, "version": __version__})
|
|
|
|
# --- публичные ссылки (без входа)
|
|
m = re.match(r"^/s/([^/.]+)\.(png|pdf)$", path)
|
|
if m:
|
|
cert = store.get_certificate_by_token(m.group(1))
|
|
if not cert:
|
|
return self._error(404, "Ссылка недействительна")
|
|
return self._send_cert_file(cert, m.group(2))
|
|
m = re.match(r"^/s/([^/]+)$", path)
|
|
if m:
|
|
cert = store.get_certificate_by_token(m.group(1))
|
|
if not cert:
|
|
return self._error(404, "Ссылка недействительна")
|
|
size = None
|
|
png = cert.get("png_path")
|
|
if png and Path(png).exists():
|
|
try:
|
|
from PIL import Image
|
|
with Image.open(png) as im:
|
|
size = im.size
|
|
except Exception:
|
|
size = None
|
|
return self._html(ui.public_cert_page_html(cert, base, size))
|
|
|
|
if path in ("/login", "/register"):
|
|
if user:
|
|
return self._redirect("/")
|
|
mode = "register" if path == "/register" else "login"
|
|
return self._html(ui.auth_page_html(mode))
|
|
|
|
if path == "/logout":
|
|
token = self._cookies().get(COOKIE_NAME)
|
|
if token:
|
|
store.delete_session(token)
|
|
return self._redirect("/login", {"Set-Cookie": self._clear_cookie()})
|
|
|
|
if path == "/":
|
|
if not user:
|
|
return self._redirect("/login")
|
|
return self._html(ui.dashboard_html(user, base))
|
|
|
|
if path == "/users":
|
|
if not user:
|
|
return self._redirect("/login")
|
|
if user["role"] != "admin":
|
|
return self._error(403, "Доступно только администратору")
|
|
return self._html(ui.users_page_html(user, base))
|
|
|
|
# --- API авторизации
|
|
if path == "/api/auth/me":
|
|
if not user:
|
|
return self._error(401, "Требуется вход")
|
|
return self._json(user)
|
|
|
|
# --- шаблоны (нужны админу для формы создания)
|
|
if path == "/api/templates":
|
|
if not self._require_api_admin():
|
|
return
|
|
return self._json(store.list_templates())
|
|
m = re.match(r"^/api/templates/([^/]+)$", path)
|
|
if m:
|
|
if not self._require_api_admin():
|
|
return
|
|
tpl = store.get_template(m.group(1))
|
|
return self._json(tpl) if tpl else self._error(404, "Шаблон не найден")
|
|
|
|
# --- пользователи (админ)
|
|
if path == "/api/users":
|
|
if not self._require_api_admin():
|
|
return
|
|
return self._json(store.list_users())
|
|
|
|
# --- сертификаты
|
|
if path == "/api/certificates":
|
|
if not user:
|
|
return self._error(401, "Требуется вход")
|
|
q = parse_qs(parsed.query)
|
|
limit = min(int(q.get("limit", ["50"])[0]), 500)
|
|
offset = max(int(q.get("offset", ["0"])[0]), 0)
|
|
recipient = None if user["role"] == "admin" else user["email"]
|
|
items = store.list_certificates(limit, offset, recipient_email=recipient)
|
|
return self._json([_public_cert(cert_api(c, base)) for c in items])
|
|
m = re.match(r"^/api/certificates/([^/]+)$", path)
|
|
if m:
|
|
if not user:
|
|
return self._error(401, "Требуется вход")
|
|
cert = self._cert_accessible(user, m.group(1))
|
|
if not cert:
|
|
return self._error(404, "Документ не найден")
|
|
return self._json(_public_cert(cert_api(cert, base)))
|
|
|
|
# --- файлы и страница документа
|
|
m = CERT_FILE_RE.match(path)
|
|
if m:
|
|
if not user:
|
|
return self._error(401, "Требуется вход")
|
|
return self._serve_file(user, m.group(1), m.group(2))
|
|
m = CERT_PAGE_RE.match(path)
|
|
if m:
|
|
if not user:
|
|
return self._redirect(f"/login?next={quote(self.path)}")
|
|
cert = self._cert_accessible(user, m.group(1))
|
|
if not cert:
|
|
return self._error(404, "Документ не найден")
|
|
return self._html(ui.cert_page_html(cert, base))
|
|
|
|
self._error(404, "Не найдено")
|
|
|
|
def _serve_file(self, user: dict, ref: str, ext: str):
|
|
cert = self._cert_accessible(user, ref)
|
|
if not cert:
|
|
return self._error(404, "Документ не найден")
|
|
return self._send_cert_file(cert, ext)
|
|
|
|
def _send_cert_file(self, cert: dict, ext: str):
|
|
path = cert.get("png_path" if ext == "png" else "pdf_path")
|
|
if not path or not Path(path).exists():
|
|
return self._error(404, "Файл ещё не готов")
|
|
data = Path(path).read_bytes()
|
|
ctype = "image/png" if ext == "png" else "application/pdf"
|
|
ascii_name = f"document-{cert['seq']}.{ext}"
|
|
utf8_name = quote(f"{cert['number']}.{ext}")
|
|
disp = f'inline; filename="{ascii_name}"; filename*=UTF-8\'\'{utf8_name}'
|
|
self._send(200, ctype, data, {"Content-Disposition": disp})
|
|
|
|
# -------------------------------------------------------------------- POST
|
|
def do_POST(self):
|
|
try:
|
|
self._route_post()
|
|
except ValueError as exc:
|
|
self._error(400, str(exc))
|
|
except Exception as exc: # noqa: BLE001
|
|
traceback.print_exc()
|
|
self._error(500, str(exc))
|
|
|
|
def _route_post(self):
|
|
parsed = urlparse(self.path)
|
|
path = unquote(parsed.path)
|
|
base = _base_url(self)
|
|
|
|
if path == "/api/auth/register":
|
|
return self._register()
|
|
if path == "/api/auth/login":
|
|
return self._login()
|
|
if path == "/api/auth/logout":
|
|
token = self._cookies().get(COOKIE_NAME)
|
|
if token:
|
|
store.delete_session(token)
|
|
return self._json({"ok": True}, extra={"Set-Cookie": self._clear_cookie()})
|
|
if path == "/api/auth/password":
|
|
return self._change_password()
|
|
|
|
user = self._current_user()
|
|
|
|
# публичная ссылка: включение/обновление токена (только админ)
|
|
m = re.match(r"^/api/certificates/([^/]+)/share$", path)
|
|
if m:
|
|
if not user or user["role"] != "admin":
|
|
return self._error(403 if user else 401,
|
|
"Доступно только администратору" if user else "Требуется вход")
|
|
cert = store.set_public_token(m.group(1), auth.new_token())
|
|
if not cert:
|
|
return self._error(404, "Документ не найден")
|
|
return self._json(_public_cert(cert_api(cert, base)))
|
|
|
|
if path in ("/api/certificates", "/api/certificates/batch", "/api/preview", "/api/templates"):
|
|
if not user or user["role"] != "admin":
|
|
return self._error(403 if user else 401,
|
|
"Доступно только администратору" if user else "Требуется вход")
|
|
|
|
if path == "/api/certificates":
|
|
body = self._read_body()
|
|
slug = (body.get("template") or "classic").strip()
|
|
template = store.get_template(slug)
|
|
if not template:
|
|
return self._error(400, f"Неизвестный шаблон: {slug}")
|
|
if not (body.get("fio") or "").strip():
|
|
return self._error(400, "Поле 'fio' обязательно")
|
|
return self._json(_public_cert(create_certificate(template, body, base, created_by=user["id"])), 201)
|
|
|
|
if path == "/api/certificates/batch":
|
|
body = self._read_body()
|
|
slug = (body.get("template") or "classic").strip()
|
|
template = store.get_template(slug)
|
|
if not template:
|
|
return self._error(400, f"Неизвестный шаблон: {slug}")
|
|
names = body.get("names") or []
|
|
if not isinstance(names, list) or not names:
|
|
return self._error(400, "Поле 'names' должно быть непустым списком")
|
|
created = []
|
|
for name in names:
|
|
item = dict(body)
|
|
item.pop("names", None)
|
|
if isinstance(name, dict):
|
|
item.update(name)
|
|
else:
|
|
item["fio"] = str(name)
|
|
created.append(_public_cert(create_certificate(template, item, base, created_by=user["id"])))
|
|
return self._json(created, 201)
|
|
|
|
if path == "/api/preview":
|
|
body = self._read_body()
|
|
slug = (body.get("template") or "classic").strip()
|
|
template = store.get_template(slug)
|
|
if not template:
|
|
return self._error(400, f"Неизвестный шаблон: {slug}")
|
|
values = build_values(template, body)
|
|
values["number"] = body.get("number") or "000000"
|
|
with tempfile.TemporaryDirectory(dir=config.TMP_DIR) as td:
|
|
tmp_png = Path(td) / "preview.png"
|
|
tmp_pdf = Path(td) / "preview.pdf"
|
|
render_certificate(template, values, tmp_png, tmp_pdf)
|
|
data = tmp_png.read_bytes()
|
|
return self._send(200, "image/png", data, {"Cache-Control": "no-store"})
|
|
|
|
if path == "/api/templates":
|
|
return self._create_template()
|
|
|
|
if path == "/api/users":
|
|
return self._create_user()
|
|
|
|
self._error(404, "Не найдено")
|
|
|
|
# ------------------------------------------------------------ операции auth
|
|
def _register(self):
|
|
body = self._read_body()
|
|
email = (body.get("email") or "").strip()
|
|
password = body.get("password") or ""
|
|
name = (body.get("name") or "").strip() or None
|
|
|
|
is_first = store.count_users() == 0
|
|
if not is_first and config.REGISTRATION != "open":
|
|
return self._error(403, "Регистрация закрыта. Обратитесь к администратору.")
|
|
|
|
err = auth.validate_registration(email, password)
|
|
if err:
|
|
return self._error(400, err)
|
|
if store.get_user_by_email(email):
|
|
return self._error(400, "Пользователь с таким email уже существует")
|
|
|
|
role = "admin" if (is_first or email.lower() == config.ADMIN_EMAIL and config.ADMIN_EMAIL) else "user"
|
|
try:
|
|
user = store.create_user(email, auth.hash_password(password), name, role)
|
|
except ValueError as exc:
|
|
return self._error(400, str(exc))
|
|
cookie = self._start_session(user)
|
|
return self._json(user, 201, {"Set-Cookie": cookie})
|
|
|
|
def _login(self):
|
|
ip = self.client_address[0]
|
|
if not _login_allowed(ip):
|
|
return self._error(429, "Слишком много попыток входа. Попробуйте позже.")
|
|
body = self._read_body()
|
|
email = (body.get("email") or "").strip()
|
|
password = body.get("password") or ""
|
|
user = store.get_user_by_email(email)
|
|
if not user or not user["is_active"] or not auth.verify_password(
|
|
password, store.get_password_hash(user["id"])):
|
|
return self._error(401, "Неверный email или пароль")
|
|
cookie = self._start_session(user)
|
|
return self._json(user, 200, {"Set-Cookie": cookie})
|
|
|
|
def _change_password(self):
|
|
user = self._require_api_user()
|
|
if not user:
|
|
return
|
|
body = self._read_body()
|
|
if not auth.verify_password(body.get("current") or "", store.get_password_hash(user["id"])):
|
|
return self._error(400, "Текущий пароль неверен")
|
|
new_password = body.get("password") or ""
|
|
if len(new_password) < auth.MIN_PASSWORD:
|
|
return self._error(400, f"Пароль должен быть не короче {auth.MIN_PASSWORD} символов")
|
|
store.update_user(user["id"], password_hash=auth.hash_password(new_password))
|
|
return self._json({"ok": True})
|
|
|
|
# ---------------------------------------------------------- операции admin
|
|
def _create_template(self):
|
|
user = self._current_user()
|
|
if not user or user["role"] != "admin":
|
|
return self._error(403, "Доступно только администратору")
|
|
body = self._read_body()
|
|
slug = (body.get("slug") or "").strip()
|
|
if not SLUG_RE.match(slug):
|
|
return self._error(400, "Некорректный slug (a-z, 0-9, _, -)")
|
|
kind = body.get("kind", "image")
|
|
if kind == "html":
|
|
html = body.get("html") or ""
|
|
if not html.strip():
|
|
return self._error(400, "Для html-шаблона нужен 'html'")
|
|
tpl = store.upsert_template(
|
|
slug=slug, title=body.get("title") or slug, kind="html",
|
|
html=html, page=body.get("page"), defaults=body.get("defaults"))
|
|
return self._json(tpl, 201)
|
|
bg_b64 = body.get("background_b64") or ""
|
|
if not bg_b64:
|
|
return self._error(400, "Для image-шаблона нужен 'background_b64'")
|
|
try:
|
|
raw = base64.b64decode(bg_b64.split(",")[-1])
|
|
except Exception:
|
|
return self._error(400, "Некорректный base64 фон")
|
|
from PIL import Image
|
|
import io
|
|
bg_dir = config.TEMPLATES_DIR / "custom" / slug
|
|
bg_dir.mkdir(parents=True, exist_ok=True)
|
|
bg_path = bg_dir / "background.png"
|
|
img = Image.open(io.BytesIO(raw)).convert("RGB")
|
|
img.save(bg_path, "PNG")
|
|
rel = str(bg_path.relative_to(config.TEMPLATES_DIR))
|
|
tpl = store.upsert_template(
|
|
slug=slug, title=body.get("title") or slug, kind="image",
|
|
background=rel, layout=body.get("layout") or [],
|
|
page=body.get("page") or {"width": img.width, "height": img.height},
|
|
defaults=body.get("defaults"))
|
|
return self._json(tpl, 201)
|
|
|
|
def _create_user(self):
|
|
body = self._read_body()
|
|
email = (body.get("email") or "").strip()
|
|
password = body.get("password") or ""
|
|
name = (body.get("name") or "").strip() or None
|
|
role = body.get("role") if body.get("role") in ("admin", "user") else "user"
|
|
err = auth.validate_registration(email, password)
|
|
if err:
|
|
return self._error(400, err)
|
|
if store.get_user_by_email(email):
|
|
return self._error(400, "Пользователь с таким email уже существует")
|
|
user = store.create_user(email, auth.hash_password(password), name, role)
|
|
return self._json(user, 201)
|
|
|
|
# ------------------------------------------------------------------- PATCH
|
|
def do_PATCH(self):
|
|
try:
|
|
parsed = urlparse(self.path)
|
|
path = unquote(parsed.path)
|
|
user = self._current_user()
|
|
if not user or user["role"] != "admin":
|
|
return self._error(403 if user else 401,
|
|
"Доступно только администратору" if user else "Требуется вход")
|
|
m = re.match(r"^/api/users/(\d+)$", path)
|
|
if not m:
|
|
return self._error(404, "Не найдено")
|
|
uid = int(m.group(1))
|
|
target = store.get_user(uid)
|
|
if not target:
|
|
return self._error(404, "Пользователь не найден")
|
|
body = self._read_body()
|
|
|
|
new_role = body.get("role")
|
|
new_active = body.get("is_active")
|
|
password = body.get("password")
|
|
|
|
# не даём остаться без администраторов
|
|
loses_admin = (target["role"] == "admin" and
|
|
((new_role is not None and new_role != "admin") or new_active is False))
|
|
if loses_admin and store.count_users(role="admin") <= 1:
|
|
return self._error(400, "Нельзя убрать последнего администратора")
|
|
if uid == user["id"] and new_active is False:
|
|
return self._error(400, "Нельзя отключить собственный аккаунт")
|
|
|
|
if new_role is not None and new_role not in ("admin", "user"):
|
|
return self._error(400, "Недопустимая роль")
|
|
if password is not None and len(password) < auth.MIN_PASSWORD:
|
|
return self._error(400, f"Пароль должен быть не короче {auth.MIN_PASSWORD} символов")
|
|
|
|
store.update_user(
|
|
uid,
|
|
role=new_role,
|
|
is_active=new_active,
|
|
name=body.get("name"),
|
|
password_hash=auth.hash_password(password) if password else None,
|
|
)
|
|
if password:
|
|
store.delete_user_sessions(uid)
|
|
return self._json(store.get_user(uid))
|
|
except Exception as exc: # noqa: BLE001
|
|
traceback.print_exc()
|
|
self._error(500, str(exc))
|
|
|
|
# ------------------------------------------------------------------ DELETE
|
|
def do_DELETE(self):
|
|
try:
|
|
parsed = urlparse(self.path)
|
|
path = unquote(parsed.path)
|
|
user = self._current_user()
|
|
if not user:
|
|
return self._error(401, "Требуется вход")
|
|
is_admin = user["role"] == "admin"
|
|
|
|
m = re.match(r"^/api/certificates/([^/]+)/share$", path)
|
|
if m:
|
|
if not is_admin:
|
|
return self._error(403, "Доступно только администратору")
|
|
cert = store.set_public_token(m.group(1), None)
|
|
if not cert:
|
|
return self._error(404, "Документ не найден")
|
|
return self._json({"public_token": None})
|
|
m = re.match(r"^/api/certificates/([^/]+)$", path)
|
|
if m:
|
|
if not is_admin:
|
|
return self._error(403, "Доступно только администратору")
|
|
ok = store.delete_certificate(m.group(1))
|
|
return self._json({"deleted": ok}, 200 if ok else 404)
|
|
m = re.match(r"^/api/templates/([^/]+)$", path)
|
|
if m:
|
|
if not is_admin:
|
|
return self._error(403, "Доступно только администратору")
|
|
ok = store.delete_template(m.group(1))
|
|
return self._json({"deleted": ok}, 200 if ok else 404)
|
|
m = re.match(r"^/api/users/(\d+)$", path)
|
|
if m:
|
|
if not is_admin:
|
|
return self._error(403, "Доступно только администратору")
|
|
uid = int(m.group(1))
|
|
if uid == user["id"]:
|
|
return self._error(400, "Нельзя удалить собственный аккаунт")
|
|
target = store.get_user(uid)
|
|
if target and target["role"] == "admin" and store.count_users(role="admin") <= 1:
|
|
return self._error(400, "Нельзя удалить последнего администратора")
|
|
ok = store.delete_user(uid)
|
|
return self._json({"deleted": ok}, 200 if ok else 404)
|
|
self._error(404, "Не найдено")
|
|
except Exception as exc: # noqa: BLE001
|
|
traceback.print_exc()
|
|
self._error(500, str(exc))
|
|
|
|
|
|
def run(host: str | None = None, port: int | None = None) -> None:
|
|
config.ensure_dirs()
|
|
store.init_db()
|
|
store.purge_expired_sessions()
|
|
host = host or config.HOST
|
|
port = port or config.PORT
|
|
httpd = ThreadingHTTPServer((host, port), Handler)
|
|
actual = httpd.server_address
|
|
users = store.count_users()
|
|
print(f"certgen: http://{host}:{actual[1]} (пользователей: {users})", flush=True)
|
|
try:
|
|
httpd.serve_forever()
|
|
except KeyboardInterrupt:
|
|
print("\ncertgen: остановлено", flush=True)
|
|
finally:
|
|
httpd.server_close()
|