fix(hub): проверка Content-Length и предел тела, тик автообновления, узкая обработка ошибок, тесты долгого claim

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_014Typ5sXWNMFirr2va2uL8o
This commit is contained in:
Andrey Epifancev 2026-10-11 22:19:54 +04:00
parent 2558248dc9
commit 34d5714898
2 changed files with 106 additions and 13 deletions

View File

@ -2,6 +2,7 @@
Маршруты /agent/* — агентский токен, остальные — клиентский (specs/hub-api/spec.md)."""
import hmac
import json
import logging
import re
import threading
import time
@ -15,6 +16,11 @@ from .commands import CommandError
from .paths import backups_dir as default_backups_dir
from .queue import ClaimedError
log = logging.getLogger(__name__)
MAX_BODY = 16 * 1024 * 1024 # предел тела запроса, байт
SOCKET_TIMEOUT = 60 # чтение запроса; ожидание долгого claim идёт уже после чтения тела и сокет не держит
class ApiError(Exception):
def __init__(self, status, error, detail=None):
@ -34,6 +40,18 @@ class Hub:
def conn(self):
return db.connect(self.db_path)
def tick(self):
"""Один шаг фонового автообновления; любой сбой логируется и не прерывает цикл."""
conn = None
try:
conn = self.conn()
queue.maybe_auto_refresh(conn)
except Exception: # noqa: BLE001
log.exception("автообновление: сбой тика")
finally:
if conn:
conn.close()
def make_backup(self, label):
conn = self.conn()
try:
@ -48,6 +66,13 @@ def _int(m, i=1):
return int(m.group(i))
def _int_param(q, name, default):
try:
return int(q.get(name, default))
except (TypeError, ValueError):
raise ApiError(400, "bad_param", f"{name}: ожидается целое число")
def _not_found(obj, what):
if obj is None:
raise ApiError(404, "not_found", what)
@ -101,7 +126,7 @@ ISSUE_FILTERS = ("sprint", "assignee", "work_kind", "fact_status", "status", "la
def h_issues(hub, conn, m, q, body):
filters = {k: q.get(k) for k in ISSUE_FILTERS}
return 200, {"issues": mirror.list_issues(conn, filters, int(q.get("limit", 50)), int(q.get("offset", 0)))}
return 200, {"issues": mirror.list_issues(conn, filters, _int_param(q, "limit", 50), _int_param(q, "offset", 0))}
def h_issue(hub, conn, m, q, body):
@ -126,7 +151,7 @@ def h_batches_post(hub, conn, m, q, body):
def h_batches_get(hub, conn, m, q, body):
return 200, {"batches": queue.list_batches(conn, int(q.get("limit", 20)))}
return 200, {"batches": queue.list_batches(conn, _int_param(q, "limit", 20))}
def h_batch_get(hub, conn, m, q, body):
@ -175,7 +200,7 @@ def h_agent_get(hub, conn, m, q, body):
def h_sync_runs(hub, conn, m, q, body):
rows = conn.execute("SELECT * FROM sync_run ORDER BY id DESC LIMIT ?", (int(q.get("limit", 20)),)).fetchall()
rows = conn.execute("SELECT * FROM sync_run ORDER BY id DESC LIMIT ?", (_int_param(q, "limit", 20),)).fetchall()
return 200, {"runs": [dict(r) for r in rows]}
@ -220,6 +245,7 @@ _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):
@ -249,7 +275,12 @@ class Handler(BaseHTTPRequestHandler):
url = urlparse(self.path)
try:
query = {k: v[0] for k, v in parse_qs(url.query).items()}
length = int(self.headers.get("Content-Length") or 0)
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:
@ -267,8 +298,6 @@ class Handler(BaseHTTPRequestHandler):
conn = self.hub.conn()
try:
status, payload = fn(self.hub, conn, match, query, body)
except (ValueError, TypeError) as e: # нечисловой limit, неверные типы полей
raise ApiError(400, "bad_request", str(e))
finally:
conn.close()
return self._send(status, payload)
@ -299,13 +328,7 @@ def serve(hub: Hub, host="0.0.0.0", port=8790) -> None:
def ticker():
while not stop.wait(60):
conn = hub.conn()
try:
queue.maybe_auto_refresh(conn)
except Exception: # noqa: BLE001
traceback.print_exc()
finally:
conn.close()
hub.tick()
threading.Thread(target=ticker, daemon=True).start()
print(f"jira-hub: http://{host}:{srv.server_port}, база {hub.db_path}")

View File

@ -1,3 +1,4 @@
import json
import threading
import pytest
@ -151,3 +152,72 @@ def test_bad_requests(hub):
urllib.request.urlopen(req)
assert e2.value.code == 400
e2.value.close()
# --- ревью Task 6 --------------------------------------------------------------
def _raw(hub, headers, path="/batches", method="POST"):
import http.client
host, port = hub["base"].removeprefix("http://").split(":")
c = http.client.HTTPConnection(host, int(port), timeout=10)
try:
c.putrequest(method, path)
c.putheader("Authorization", "Bearer " + hub["client"].token)
for k, v in headers.items():
c.putheader(k, v)
c.endheaders()
r = c.getresponse()
return r.status, json.loads(r.read().decode("utf-8"))
finally:
c.close()
@pytest.mark.parametrize("value,status,error", [("-5", 400, "bad_content_length"), ("abc", 400, "bad_content_length"),
(str(api.MAX_BODY + 1), 413, "too_large")])
def test_content_length_checked(hub, value, status, error):
got, payload = _raw(hub, {"Content-Length": value})
assert got == status and payload["error"] == error
def test_numeric_param_400_internal_bug_500(hub, monkeypatch):
with pytest.raises(HubError) as e:
hub["client"].get("/batches", limit="x")
assert e.value.status == 400
def boom(*a):
raise TypeError("внутренний баг")
monkeypatch.setattr(api, "_COMPILED", [("GET", api.re.compile(r"^/issues$"), boom, "client")])
with pytest.raises(HubError) as e:
hub["client"].get("/issues")
assert e.value.status == 500 and e.value.error == "internal"
with pytest.raises(HubError) as e: # сервер жив
hub["client"].get("/issues")
assert e.value.status == 500
def test_tick_survives_connection_failure(hub, monkeypatch):
def broken():
raise OSError("диск недоступен")
monkeypatch.setattr(hub["hub"], "conn", broken)
hub["hub"].tick() # не бросает
monkeypatch.undo()
hub["hub"].tick()
def test_claim_long_poll_returns_new_command(hub):
import time
hub["hub"].claim_wait = 4
threading.Timer(0.5, lambda: hub["client"].post("/batches", {"title": "x", "commands": [
{"type": "ensure_fields", "params": {"key": "A-1", "duedate": "2026-10-23"}}]})).start()
t0 = time.monotonic()
cmd = hub["agent"].post("/agent/claim", {})["command"]
assert cmd and cmd["type"] == "ensure_fields" and time.monotonic() - t0 < 4
def test_claim_long_poll_times_out_with_null(hub):
import time
hub["hub"].claim_wait = 1.5
t0 = time.monotonic()
assert hub["agent"].post("/agent/claim", {})["command"] is None
assert 1.0 <= time.monotonic() - t0 < 4