"""Реле через общую папку (specs/hub-api, «Реле через общую папку»): файлы-запросы агента из /req/ пересылаются в HTTP этого же хаба, ответы кладутся в /resp/. HTTP API, обработчики и токены не меняются; реле для них — ещё один клиент. Нужно, когда VPN на виртуалке закрывает сеть до хаба (design.md folder-transport).""" import http.client import json import os import threading import time import traceback import urllib.error import urllib.request from datetime import datetime, timezone from pathlib import Path from . import settings CLAIM_PATH = "/agent/claim" CLAIM_MIN_LEFT = 35 # с: claim пересылается, только если до deadline осталось не меньше (хаб держит claim ≤ 30 с) SETTINGS_EVERY = 5 # с: как часто перечитывать relay_dir CLEANUP_EVERY = 60 # с: как часто убирать мусор RESP_TTL = 600 # с: ответ, за которым агент не пришёл REQ_GRACE = 60 # с после deadline: запрос, который никто не взял TMP_TTL = 600 # с: брошенные временные файлы (имя на точку) BAD_JSON_WAIT = 2 # с: сколько перечитывать нечитаемый запрос FORWARD_TIMEOUT = 90 # с: ожидание ответа своего HTTP (claim держит до 30 с) METHODS = ("GET", "POST", "PUT") def say(msg: str) -> None: print(f"[{time.strftime('%d/%b/%Y %H:%M:%S')}] {msg}", flush=True) def parse_deadline(value) -> float | None: """ISO 8601 → секунды эпохи; без зоны считается UTC; мусор → None.""" if not isinstance(value, str): return None try: dt = datetime.fromisoformat(value) except ValueError: return None if dt.tzinfo is None: dt = dt.replace(tzinfo=timezone.utc) return dt.timestamp() def write_atomic(directory: Path, name: str, payload: dict) -> None: """Записать JSON под временным именем на точку и переименовать: читатель видит файл только целиком.""" tmp = Path(directory) / f".{name}.tmp" with open(tmp, "w", encoding="utf-8") as f: json.dump(payload, f, ensure_ascii=False) f.flush() os.fsync(f.fileno()) os.replace(tmp, Path(directory) / name) class Relay: def __init__(self, hub, port: int, host: str = "127.0.0.1", poll: float = 0.1): self.hub, self.port, self.host, self.poll = hub, port, host, poll self.root: Path | None = None self._settings_at: float | None = None self._cleanup_at = 0.0 self._ready = False self._warned: str | None = None self._cleaned: set[Path] = set() # папки (по resolve()), чей work/ вычищен после старта реле self.workers: list[threading.Thread] = [] # --- цикл --------------------------------------------------------------------- def run(self, stop: threading.Event) -> None: while not stop.is_set(): try: self.step() except Exception: # noqa: BLE001 — сбой одного шага не останавливает реле traceback.print_exc() stop.wait(self.poll) def step(self) -> None: now = time.time() if self._settings_at is None or now - self._settings_at >= SETTINGS_EVERY: self._settings_at = now self._reload() if self.root is None: return try: if not self.root.is_dir(): self._warn(f"relay: папки {self.root} нет — жду") self._ready = False return if not self._ready: self._prepare() for path in sorted((self.root / "req").iterdir()): if path.name.startswith(".") or not path.name.endswith(".json"): continue self._take(path) except OSError as e: # нет доступа (TCC), подпапку удалили — подготовить заново на следующем шаге self._ready = False self._warn(f"relay: нет доступа к папке {self.root} — {e.strerror or e}; жду") return if now - self._cleanup_at >= CLEANUP_EVERY: self._cleanup_at = now self.cleanup(now) def _warn(self, msg: str) -> None: """Одно предупреждение на причину: без потока одинаковых строк каждые 0,1 с.""" if msg != self._warned: say(msg) self._warned = msg def join(self, timeout: float = 5) -> None: for w in list(self.workers): w.join(timeout) # --- папка ---------------------------------------------------------------------- def _reload(self) -> None: conn = self.hub.conn() try: value = settings.get(conn, "relay_dir") finally: conn.close() root = Path(value) if value else None if root != self.root: self.root, self._ready, self._warned = root, False, None say(f"relay: папка {root}" if root else "relay: выключено") def _prepare(self) -> None: for sub in ("req", "resp", "work"): (self.root / sub).mkdir(exist_ok=True) key = _same_dir_key(self.root) if key not in self._cleaned: # первая подготовка этой папки после старта: взятые прошлым процессом запросы уже никто не обработает for stale in (self.root / "work").iterdir(): stale.unlink(missing_ok=True) say(f"relay: удалён незавершённый запрос {stale.name} (старт реле)") self._cleaned.add(key) else: # пропал подкаталог или не было доступа: взятые запросы ещё обрабатываются своими потоками — work/ не трогаем say(f"relay: папка {self.root} подготовлена заново") self._ready, self._warned = True, None def _take(self, path: Path) -> None: target = self.root / "work" / path.name try: os.rename(path, target) except FileNotFoundError: if path.exists(): # пропала work/ — step() подготовит папку заново raise return # агент успел удалить свой просроченный запрос worker = threading.Thread(target=self.process, args=(self.root, target), daemon=True) worker.start() self.workers = [w for w in self.workers if w.is_alive()] + [worker] # --- обработка запроса ------------------------------------------------------------- def process(self, root: Path, work: Path) -> None: started = time.monotonic() try: req = self._read(work) if req is None: say(f"relay: нечитаемый запрос {work.name} — выброшен") return method, path = req.get("method"), req.get("path") deadline = parse_deadline(req.get("deadline")) left = deadline - time.time() if deadline is not None else -1.0 if left <= 0 or (method == "POST" and path == CLAIM_PATH and left < CLAIM_MIN_LEFT): say(f"relay {method} {path} просрочен (осталось {left:.0f} с) — выброшен") return (root / "resp").mkdir(exist_ok=True) # до пересылки: выданный claim не должен уйти в никуда status, body = self.forward(method, path, req.get("body"), req.get("authorization")) write_atomic(root / "resp", work.name, {"id": req.get("id"), "status": status, "body": body}) say(f"relay {method} {path} {status} {time.monotonic() - started:.1f} с") except OSError as e: # папка пропала на ходу — агент дождётся своего срока say(f"relay: сбой файла {work.name}: {e}") finally: work.unlink(missing_ok=True) def _read(self, work: Path) -> dict | None: end = time.monotonic() + BAD_JSON_WAIT while True: try: with open(work, encoding="utf-8") as f: data = json.load(f) return data if isinstance(data, dict) else None except (ValueError, OSError): if time.monotonic() >= end: return None time.sleep(0.1) def forward(self, method, path, body, authorization) -> tuple[int, str]: if method not in METHODS or not isinstance(path, str) or not path.startswith("/"): return 400, _error("bad_request", f"неверный метод или путь: {method} {path}") data = json.dumps(body, ensure_ascii=False).encode("utf-8") if body is not None else None try: req = urllib.request.Request(f"http://{self.host}:{self.port}{path}", data=data, method=method) if authorization: req.add_header("Authorization", str(authorization)) if data is not None: req.add_header("Content-Type", "application/json; charset=utf-8") with urllib.request.urlopen(req, timeout=FORWARD_TIMEOUT) as resp: return resp.status, resp.read().decode("utf-8", "replace") except urllib.error.HTTPError as e: try: return e.code, e.read().decode("utf-8", "replace") finally: e.close() except ValueError as e: # недопустимый заголовок (перевод строки) или URL return 400, _error("bad_request", str(e)) except (urllib.error.URLError, http.client.HTTPException, OSError) as e: return 502, _error("relay", f"хаб недоступен изнутри: {e}") # --- уборка ------------------------------------------------------------------------ def cleanup(self, now: float) -> None: if self.root is None or not self.root.is_dir(): return for p in _entries(self.root / "resp"): ttl = TMP_TTL if p.name.startswith(".") else RESP_TTL if _age(p, now) > ttl: p.unlink(missing_ok=True) for p in _entries(self.root / "req"): if p.name.startswith("."): if _age(p, now) > TMP_TTL: p.unlink(missing_ok=True) continue try: deadline = parse_deadline(json.loads(p.read_text(encoding="utf-8")).get("deadline")) except (OSError, ValueError, AttributeError): deadline = None if deadline is None or now - deadline > REQ_GRACE: p.unlink(missing_ok=True) def _same_dir_key(path: Path) -> Path: """Ключ папки для «уже вычищена»: resolve() — симлинк и другая запись той же папки совпадают (minor-1 #16, решение 12); ошибка resolve — путь как записан.""" try: return path.resolve() except (OSError, RuntimeError): return path def _error(error: str, detail: str) -> str: return json.dumps({"error": error, "detail": detail}, ensure_ascii=False) def _entries(directory: Path) -> list[Path]: try: return list(directory.iterdir()) except FileNotFoundError: return [] def _age(p: Path, now: float) -> float: try: return now - p.stat().st_mtime except FileNotFoundError: return 0.0