Andrey Epifancev 2673bac5d8 docs: ядро и модули хаба; тест границы ядра (#87)
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
2026-10-23 23:20:06 +04:00

500 lines
25 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

"""HTTP API хаба (stdlib http.server, поток на запрос, соединение SQLite на запрос).
Маршруты /agent/* — агентский токен, остальные — клиентский (specs/hub-api/spec.md).
Без токена — статика без данных: веб-приложение (GET /app…) и перенаправление старых адресов доски (GET /, GET /links)."""
import base64
import hmac
import json
import logging
import re
import threading
import time
import traceback
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from urllib.parse import parse_qs, unquote, urlparse
from . import backup, db, files, projects, queue, registry, relay, release, settings
from .errors import CommandError, HttpMapped
from .http import ApiError, Raw, int_group, int_param, mapped, not_found
from .paths import backups_dir as default_backups_dir
from .paths import secure_dir
from .queue import ClaimedError
log = logging.getLogger(__name__)
MAX_BODY = 16 * 1024 * 1024 # предел тела запроса, байт
WEB_APP = Path(__file__).resolve().parent.parent / "web" / "app" # сборка веб-приложения (make web, не в git); читается на каждый запрос
ASSET_TYPES = {".js": "text/javascript; charset=utf-8", ".css": "text/css; charset=utf-8", ".svg": "image/svg+xml",
".png": "image/png", ".ico": "image/x-icon", ".woff2": "font/woff2", ".map": "application/json"}
ASSET_CACHE = {"Cache-Control": "public, max-age=31536000, immutable"} # имена файлов сборки содержат хэш (R-3)
NOT_BUILT = ("<!doctype html><meta charset=\"utf-8\"><title>Хаб Jira</title><p>Веб-приложение не собрано: выполните "
"<code>make web</code> в каталоге jira-hub и перезапустите хаб.</p>").encode("utf-8")
SOCKET_TIMEOUT = 60 # чтение запроса; ожидание долгого claim идёт уже после чтения тела и сокет не держит
SWEEP_EVERY = 24 * 3600 # уборка files/ не чаще раза в сутки (hub-api «Хранилище файлов вложений»)
class Hub:
def __init__(self, db_path, backups_dir=None, release_dir=None, web_dir=None, files_dir=None):
self.db_path = Path(db_path)
self.web_dir = Path(web_dir) if web_dir else WEB_APP
self.backups_dir = Path(backups_dir) if backups_dir else default_backups_dir()
self.release_dir = Path(release_dir) if release_dir else self.db_path.parent / "release"
self.files_dir = Path(files_dir) if files_dir else self.db_path.parent / "files"
self.last_sweep = None # время прошлой уборки files/, в памяти (решение 15)
self.module_state: dict[str, dict] = {} # память модулей между тиками (имя модуля → словарь)
# последнее из claim; в памяти (R-3). reported — был ли claim с запуска хаба: до него launcher_version
# неизвестен, а не null (hub-hygiene-d, решение 7)
self.agent_update = {"update_state": None, "launcher_version": None, "reported": False}
self.claim_wait = 30
secure_dir(self.backups_dir) # каталог копий закрыт при каждом старте, а не только после первой копии
secure_dir(self.files_dir) # files/ — 0700 при каждом старте, как backups/
conn = self.conn()
settings.ensure_defaults(conn)
conn.close()
def conn(self):
return db.connect(self.db_path)
def tick(self):
"""Один шаг фоновых работ: уборка файлов ядра (не чаще раза в сутки), затем tick модулей. Сбой каждой
работы логируется и не мешает ни другой, ни следующему тику; соединение открывается при первой работе."""
works = [("уборка файлов: сбой", self.sweep_files)]
for m in registry.modules():
for fn in m.tick:
label = getattr(fn, "label", getattr(fn, "__name__", m.name))
works.append((f"{label}: сбой тика", lambda conn, fn=fn: fn(self, conn)))
conn = None
try:
for message, work in works:
try:
if conn is None:
conn = self.conn()
work(conn)
except Exception: # noqa: BLE001
log.exception(message)
finally:
if conn is not None:
conn.close()
def sweep_files(self, conn, now=None):
"""Уборка files/ (files.sweep) не чаще раза в SWEEP_EVERY; время прошлой — в памяти процесса, после
перезапуска хаба первая уборка — на первом тике (решение 15). None — ещё рано."""
now = time.time() if now is None else now
if self.last_sweep is not None and now - self.last_sweep < SWEEP_EVERY:
return None
self.last_sweep = now
removed = files.sweep(conn, self.files_dir, now)
if removed:
log.info("уборка файлов: удалено %d", len(removed))
return removed
def note_agent_update(self, body):
"""update_state и launcher_version из claim: хранятся в памяти, агент шлёт их в каждом claim (R-3).
Значение неверного типа игнорируется — claim из-за него не отвергается; сам claim отмечается (reported)."""
new = dict(self.agent_update)
new["reported"] = True
us = body.get("update_state")
if us is None or isinstance(us, dict):
if us is not None and isinstance(us.get("error"), str):
us = {**us, "error": us["error"][:500]}
new["update_state"] = us
lv = body.get("launcher_version")
if lv is None or (isinstance(lv, int) and not isinstance(lv, bool)):
new["launcher_version"] = lv
self.agent_update = new
def make_backup(self, label):
conn = self.conn()
try:
backup.make_backup(conn, self.backups_dir, label, int(settings.get(conn, "backup_keep") or 30))
finally:
conn.close()
# --- обработчики: fn(hub, conn, match, query, body) -> (status, payload) ----------
def h_claim(hub, conn, m, q, body):
hub.note_agent_update(body)
deadline = time.monotonic() + hub.claim_wait
while True:
cmd = queue.claim(conn, body.get("agent_version"), body.get("settings_version"), body.get("last_jira_call_at"),
backup=hub.make_backup, files_dir=hub.files_dir)
if cmd or time.monotonic() >= deadline:
break
time.sleep(1)
try: # команда уже выдана: публикация не должна превращать ответ в 500 (I-1)
rel = release.current(hub.release_dir)
except Exception: # noqa: BLE001
log.exception("release: чтение публикации при claim")
rel = None
return 200, {"command": cmd, "settings_version": settings.get(conn, "settings_version"),
"agent_release": {"version": rel["version"], "sha256": rel["sha256"]} if rel else None}
def h_agent_release(hub, conn, m, q, body):
rel = release.load(hub.release_dir)
if rel is None:
raise ApiError(404, "no_release", "версия агента не опубликована")
return 200, rel
def h_agent_file(hub, conn, m, q, body):
"""Файл вложения агенту: {file, size, content}, content — base64 (проходит и реле папки, и HTTP). id не из 64 hex
нижнего регистра — тот же 404 no_file (решение 16)."""
data = files.read(hub.files_dir, m.group(1))
if data is None:
raise ApiError(404, "no_file", f"файла {m.group(1)} нет в хранилище хаба")
return 200, {"file": m.group(1), "size": len(data), "content": base64.b64encode(data).decode("ascii")}
def h_agent_settings(hub, conn, m, q, body):
return 200, settings.for_agent(conn)
def h_agent_command(hub, conn, m, q, body):
"""Команда по id для восстановления из журнала агента: параметры — как при выдаче (с size у команд с файлом)."""
cmd = not_found(queue.get_command(conn, int_group(m)), "command")
return 200, {**cmd, "params": queue.deliver_params(conn, cmd, hub.files_dir)}
def h_agent_result(hub, conn, m, q, body):
try:
cmd = queue.report_result(conn, int_group(m), body.get("status"), body.get("result"), body.get("error"),
body.get("jira_calls", 0), body.get("rate_limited", 0), files_dir=hub.files_dir)
except KeyError:
raise ApiError(404, "not_found", "command")
except queue.NotClaimedError as e:
raise ApiError(409, "not_claimed", str(e))
except ValueError as e:
raise ApiError(422, "bad_status", str(e))
return 200, cmd
def h_app(hub, conn, m, q, body):
"""Веб-приложение: файлы assets/ сборки как есть, любой другой путь /app… — index.html (маршруты на клиенте).
Путь файла после разрешения обязан лежать внутри assets/, иначе 404; без сборки — 503 с подсказкой."""
index = hub.web_dir / "index.html"
if not index.is_file():
return 503, Raw("text/html; charset=utf-8", NOT_BUILT)
rest = unquote(m.group(1) or "")
if rest.startswith("assets/"):
if "\x00" in rest: # T1.1: нулевой байт — 404, а не 500 из resolve()
raise ApiError(404, "not_found", "/app/" + rest.replace("\x00", "%00"))
assets = (hub.web_dir / "assets").resolve()
try:
f = (hub.web_dir / rest).resolve()
except (OSError, ValueError):
raise ApiError(404, "not_found", "/app/" + rest) from None
if not f.is_relative_to(assets) or not f.is_file():
raise ApiError(404, "not_found", "/app/" + rest)
return 200, Raw(ASSET_TYPES.get(f.suffix, "application/octet-stream"), f.read_bytes(), ASSET_CACHE)
return 200, Raw("text/html; charset=utf-8", index.read_bytes())
def h_root(hub, conn, m, q, body):
"""Старые адреса доски (/, /links) — на страницу доски веб-приложения."""
return 302, Raw("text/html; charset=utf-8", b"", {"Location": "/app/board", "Cache-Control": "no-cache"})
def h_batches_post(hub, conn, m, q, body):
if not body.get("title") or not isinstance(body["title"], str):
raise ApiError(422, "bad_batch", "title — непустая строка")
if not isinstance(body.get("author") or "api", str):
raise ApiError(422, "bad_batch", "author — строка")
if not isinstance(body.get("commands"), list):
raise ApiError(422, "bad_batch", "commands — список команд")
if body.get("project") is not None and not isinstance(body["project"], str):
raise ApiError(422, "bad_batch", "project — строка")
if body.get("project") == "":
raise ApiError(422, "bad_batch", "project: пустая строка")
try:
return 201, queue.create_batch(conn, body["title"], body.get("author") or "api", body.get("commands"),
files_dir=hub.files_dir, project=body.get("project"))
except HttpMapped as e: # раньше CommandError: исключения модулей несут свой код (M-2)
raise mapped(e) from None
except CommandError as e:
raise ApiError(422, "bad_batch", str(e))
def h_batches_get(hub, conn, m, q, body):
return 200, {"batches": queue.list_batches(conn, int_param(q, "limit", 20), project=q.get("project") or None)}
def h_batch_get(hub, conn, m, q, body):
return 200, not_found(queue.batch_view(conn, int_group(m)), "batch")
def h_batch_cancel(hub, conn, m, q, body):
try:
return 200, queue.cancel_batch(conn, int_group(m))
except KeyError:
raise ApiError(404, "not_found", "batch")
def _put_file(hub, body, allow_empty=False):
"""Общая часть POST /files и POST /agent/files: {content: base64} → {file: sha256, size}. Пробельные символы
(переносы строк base64) отбрасываются до строгой проверки (решение 14)."""
content = body.get("content")
if not isinstance(content, str):
raise ApiError(400, "bad_request", "content — строка base64")
try:
data = base64.b64decode("".join(content.split()), validate=True)
except ValueError: # binascii.Error — подкласс ValueError
raise ApiError(400, "bad_request", "content — не base64") from None
try:
sha, size = files.put(hub.files_dir, data, allow_empty=allow_empty)
except files.EmptyFile:
raise ApiError(422, "empty_file", "пустой файл") from None
except files.TooLarge:
raise ApiError(413, "too_large", f"файл больше {files.MAX_SIZE} байт ({len(data)})") from None
return 200, {"file": sha, "size": size}
def h_files_post(hub, conn, m, q, body):
return _put_file(hub, body)
def h_agent_files_post(hub, conn, m, q, body):
"""Файл от агента (ответ на запрос файла, решение 5): как POST /files, но пустой файл принимается (решение 13)."""
return _put_file(hub, body, allow_empty=True)
def h_command_retry(hub, conn, m, q, body):
if not isinstance(body.get("author") or "api", str):
raise ApiError(422, "bad_retry", "author — строка")
if body.get("project") is not None and not isinstance(body["project"], str):
raise ApiError(422, "bad_retry", "project — строка")
if body.get("project") == "":
raise ApiError(422, "bad_retry", "project: пустая строка")
try:
return 201, queue.retry_command(conn, int_group(m), body.get("author") or "api", files_dir=hub.files_dir,
project=body.get("project"))
except KeyError:
raise ApiError(404, "not_found", "command")
except HttpMapped as e: # раньше CommandError: исключения модулей несут свой код (M-2)
raise mapped(e) from None
except CommandError as e:
raise ApiError(422, "bad_batch", str(e))
def h_command_cancel(hub, conn, m, q, body):
try:
return 200, queue.cancel_command(conn, int_group(m), bool(body.get("force")))
except KeyError:
raise ApiError(404, "not_found", "command")
except ClaimedError as e:
raise ApiError(409, "claimed", e.claimed_at)
def h_agent_get(hub, conn, m, q, body):
view = queue.agent_view(conn)
rel = release.current(hub.release_dir)
au = hub.agent_update
us, lv, reported = au["update_state"], au["launcher_version"], au["reported"]
view.update(release=rel, update_state=us, launcher_version=lv, agent_reported=reported,
update_hint=release.hint(rel, view, us, lv, reported=reported))
return 200, view
def _settings_view(conn) -> dict:
"""Настройки ядра и модулей плюс вычисляемые поля модулей (settings_view)."""
out = settings.get_all(conn)
for mod in registry.modules():
if mod.settings_view:
out.update(mod.settings_view(conn))
return out
def h_settings_get(hub, conn, m, q, body):
return 200, _settings_view(conn)
def h_settings_put(hub, conn, m, q, body):
"""Записать настройки пакетом (всё или ничего). До записи проверяются защищённые ключи и значения
(settings.validate): ошибка — 422, версия настроек не меняется. Ответ — все настройки после записи, как у
GET /settings (hub-hygiene-d, решение 9)."""
bad = [k for k in body if k in settings.PROTECTED]
if bad:
raise ApiError(422, "protected", ", ".join(bad))
for key, value in body.items():
try:
settings.validate(key, value)
except ValueError as e:
raise ApiError(422, "bad_setting", str(e))
with db.transaction(conn):
for key, value in body.items():
settings.set_(conn, key, value)
return 200, _settings_view(conn)
def h_projects_get(hub, conn, m, q, body):
return 200, {"projects": projects.list_projects(conn)}
def h_projects_post(hub, conn, m, q, body):
"""Новый проект: проект и пакеты, которые ставят модули (on_project_created; у jira — «Новый проект <метка>») —
в одной транзакции; ошибка любого шага не оставляет ни проекта, ни пакета (hub-api, «Маршруты проектов»)."""
author = body.get("author")
if author is None or author == "":
author = "api"
elif not isinstance(author, str):
raise ApiError(422, "bad_project", "author — строка")
batch_id = None
try:
with db.transaction(conn):
project = projects.create(conn, body.get("label"), body.get("name"), body.get("board_id"))
for mod in registry.modules():
batch = mod.on_project_created(conn, project, author) if mod.on_project_created else None
if batch is not None and batch_id is None:
batch_id = batch["id"]
except projects.ProjectExists as e:
raise ApiError(409, "project_exists", str(e))
except projects.ProjectError as e:
raise ApiError(422, "bad_project", str(e))
except CommandError as e:
raise ApiError(422, "bad_batch", str(e))
return 201, {"project": project, "batch_id": batch_id}
CORE_ROUTES = [
("GET", r"^/(links)?$", h_root, None),
("GET", r"^/app(?:/(.*))?$", h_app, None),
("POST", r"^/agent/claim$", h_claim, "agent"),
("GET", r"^/agent/settings$", h_agent_settings, "agent"),
("GET", r"^/agent/commands/(\d+)$", h_agent_command, "agent"),
("POST", r"^/agent/commands/(\d+)/result$", h_agent_result, "agent"),
("GET", r"^/agent/release$", h_agent_release, "agent"),
("GET", r"^/agent/files/([^/]+)$", h_agent_file, "agent"),
("POST", r"^/agent/files$", h_agent_files_post, "agent"),
("POST", r"^/batches$", h_batches_post, "client"),
("GET", r"^/batches$", h_batches_get, "client"),
("GET", r"^/batches/(\d+)$", h_batch_get, "client"),
("POST", r"^/batches/(\d+)/cancel$", h_batch_cancel, "client"),
("POST", r"^/files$", h_files_post, "client"),
("POST", r"^/commands/(\d+)/retry$", h_command_retry, "client"),
("POST", r"^/commands/(\d+)/cancel$", h_command_cancel, "client"),
("GET", r"^/agent$", h_agent_get, "client"),
("GET", r"^/settings$", h_settings_get, "client"),
("PUT", r"^/settings$", h_settings_put, "client"),
("GET", r"^/projects$", h_projects_get, "client"),
("POST", r"^/projects$", h_projects_post, "client"),
]
# маршрут ядра и модуля не делят метод + шаблон: пересечение — ошибка старта; ROUTES собирается после проверки
registry.check(registry.modules(), CORE_ROUTES, settings.CORE_DEFAULTS)
ROUTES = CORE_ROUTES + [r for mod in registry.modules() for r in mod.routes]
_COMPILED = [(m, re.compile(p), fn, scope) for m, p, fn, scope in ROUTES]
class Handler(BaseHTTPRequestHandler):
timeout = SOCKET_TIMEOUT
hub: Hub = None # подставляется в make_server
def do_GET(self):
self._dispatch("GET")
def do_POST(self):
self._dispatch("POST")
def do_PUT(self):
self._dispatch("PUT")
def log_message(self, fmt, *args): # тише стандартного: без адреса клиента
print(f"[{self.log_date_time_string()}] {fmt % args}")
def _auth(self, scope):
token = (self.headers.get("Authorization") or "").removeprefix("Bearer ").strip()
conn = self.hub.conn()
try:
tokens = settings.tokens(conn)
finally:
conn.close()
expected = tokens["agent" if scope == "agent" else "client"]
if not token or not hmac.compare_digest(token.encode("utf-8"), expected.encode("utf-8")):
raise ApiError(401, "unauthorized", "неверный токен для этого маршрута")
def _dispatch(self, method):
url = urlparse(self.path)
try:
query = {k: v[0] for k, v in parse_qs(url.query).items()}
raw_len = (self.headers.get("Content-Length") or "0").strip()
if not raw_len.isascii() or not raw_len.isdigit():
raise ApiError(400, "bad_content_length", "Content-Length должен быть неотрицательным целым")
length = int(raw_len)
if length > MAX_BODY:
raise ApiError(413, "too_large", f"тело больше {MAX_BODY} байт")
raw = self.rfile.read(length) if length else b""
for route_method, rx, fn, scope in _COMPILED:
if route_method != method:
continue
match = rx.match(url.path)
if not match:
continue
if scope is not None:
self._auth(scope)
try:
body = json.loads(raw.decode("utf-8")) if raw else {}
except ValueError as e: # битый JSON или не UTF-8
raise ApiError(400, "bad_json", str(e))
if not isinstance(body, dict):
raise ApiError(400, "bad_json", "ожидается объект")
conn = self.hub.conn()
try:
status, payload = fn(self.hub, conn, match, query, body)
finally:
conn.close()
return self._send(status, payload)
raise ApiError(404, "not_found", url.path)
except ApiError as e:
self._send(e.status, {"error": e.error, "detail": e.detail})
except Exception as e: # noqa: BLE001 — любой сбой обработчика виден клиенту и в логе
traceback.print_exc()
self._send(500, {"error": "internal", "detail": str(e)})
def _send(self, status, payload):
if isinstance(payload, Raw):
ctype, data = payload.content_type, payload.data
extra = payload.headers if payload.headers is not None else {"Cache-Control": "no-cache"}
else:
ctype, extra = "application/json; charset=utf-8", {}
data = json.dumps(payload, ensure_ascii=False, default=str).encode("utf-8")
self.send_response(status)
self.send_header("Content-Type", ctype)
self.send_header("Content-Length", str(len(data)))
for name, value in extra.items():
self.send_header(name, value)
self.end_headers()
self.wfile.write(data)
def make_server(hub: Hub, host="0.0.0.0", port=0) -> ThreadingHTTPServer:
handler = type("HubHandler", (Handler,), {"hub": hub})
return ThreadingHTTPServer((host, port), handler)
def start(hub: Hub, host="0.0.0.0", port=8790) -> tuple[ThreadingHTTPServer, threading.Event]:
"""Создать HTTP-сервер и запустить в фоне тикер автообновления и реле общей папки.
Обслуживание HTTP (serve_forever) запускает вызывающий; stop останавливает тикер и реле."""
srv = make_server(hub, host, port)
stop = threading.Event()
def ticker():
while not stop.wait(60):
hub.tick()
threading.Thread(target=ticker, daemon=True).start()
target = "127.0.0.1" if host in ("", "0.0.0.0") else host
threading.Thread(target=relay.Relay(hub, srv.server_port, host=target).run, args=(stop,), daemon=True).start()
return srv, stop
def serve(hub: Hub, host="0.0.0.0", port=8790) -> None:
srv, stop = start(hub, host, port)
print(f"jira-hub: http://{host}:{srv.server_port}, база {hub.db_path}", flush=True)
try:
srv.serve_forever()
except KeyboardInterrupt:
pass
finally:
stop.set()
srv.server_close()