From eac8352dc2340e474a8e50352f85fae6a5ca098a Mon Sep 17 00:00:00 2001 From: ochenstarik-ui <267932263+ochenstarik-ui@users.noreply.github.com> Date: Thu, 3 Sep 2026 01:16:43 +0700 Subject: [PATCH] =?UTF-8?q?fix(release,web):=20=D0=B2=D0=BE=D1=80=D0=BE?= =?UTF-8?q?=D1=82=D0=B0=20=D0=BF=D1=83=D0=B1=D0=BB=D0=B8=D0=BA=D0=B0=D1=86?= =?UTF-8?q?=D0=B8=D0=B8=20=D0=BF=D0=B5=D1=80=D0=B5=D1=81=D1=82=D0=B0=D0=BB?= =?UTF-8?q?=D0=B8=20=D0=BF=D1=80=D0=BE=D0=BF=D1=83=D1=81=D0=BA=D0=B0=D1=82?= =?UTF-8?q?=D1=8C=20=D0=B2=D1=81=D1=91=20=D0=BF=D0=BE=D0=B4=D1=80=D1=8F?= =?UTF-8?q?=D0=B4,=20/api/action=20=E2=80=94=20=D0=BC=D0=B5=D0=B6=D1=81?= =?UTF-8?q?=D0=B0=D0=B9=D1=82=D0=BE=D0=B2=D1=8B=D0=B5=20=D0=B7=D0=B0=D0=BF?= =?UTF-8?q?=D1=80=D0=BE=D1=81=D1=8B?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit P0 аудита, каждый сначала подтверждён исполнением, а не принят со слов. 1. Release Gate объявлял проверку хеша, которой не было. Печаталась строка PACKAGE_HASH_VERIFIED=True при том, что hashlib в scripts/release_gate.py не вызывался ни разу: скачивались байты 0-10 через заголовок Range, и этого хватало, чтобы счесть хеш проверенным. «Проверенным ассетом» при этом оказывался первый в списке — checksums.txt, а не пакет. 2. Ворота публикации были fail-open. Измерено в трёх условиях: полный обрыв сети -> PASS, манифест 404 -> PASS, пакет 404 -> PASS. Ворота пропускали релиз при любом исходе, включая полное отсутствие релиза. Разделено на офлайновую часть (проверки 1-7: версии, тесты, updater, статика, секреты, список разрешённых адресов) и Publication Gate: релиз есть, ассеты есть, пакет скачан ЦЕЛИКОМ, SHA-256 сошёлся с опубликованным checksums.txt. Публикационные ворота блокируют в режиме публикации (--publication или HERMES_RELEASE_PUBLICATION_GATE=1); в обычном прогоне CI, где релиза для ветки нет и быть не должно, результат сообщается как есть и не блокирует. Неизмеренное называется причиной, а не выдаётся за проверенное. Проверено на живом релизе v0.1.3-b1: два пакета скачаны целиком, хеши сошлись. 3. POST /api/action на loopback принимал межсайтовые запросы. Токен там не требуется, а действие меняет состояние: удаляет учётные данные, чистит аккаунты, переключает маршрутизацию, запускает входы OAuth. CORS от этого не защищает — он мешает прочитать ответ, а не отправить запрос. Измерено на конфигурации по умолчанию: POST с Content-Type text/plain уходит кросс-сайтом без предварительного запроса, request.json() разбирает тело независимо от Content-Type, и запрос с Origin чужого сайта без токена доходил до исполнителя действий. Проверяется Sec-Fetch-Site, при его отсутствии — Origin против адреса запроса. Собственный интерфейс, адресная строка и не-браузерные клиенты работают как раньше. Защита распространена на все пять небезопасных методов, не только на /api/action. 4. pricing fallback: safe_load вместо safe_dump. dump сериализовал текст обратно в строку, проверка isinstance(data, dict) не выполнялась никогда, таблица цен не загружалась ни разу, а except это глушил. 5. Симуляция Linux в тесте stop_running_hub падала на Windows: os.getuid там не существует. Тесты: 756 -> 776 passed, 2 skipped, 4 deselected. ruff check . чисто. Co-Authored-By: Claude Opus 5 --- scripts/release_gate.py | 230 ++++++++++++------ .../router/telemetry_service.py | 6 +- src/antigravity_provider/router/web/server.py | 88 ++++++- tests/test_a37_isolation_guards.py | 67 +++++ tests/test_a59_visible_update.py | 5 +- tests/test_p0_release_gate.py | 160 ++++++++++++ 6 files changed, 480 insertions(+), 76 deletions(-) diff --git a/scripts/release_gate.py b/scripts/release_gate.py index 2ec0460..7530f31 100644 --- a/scripts/release_gate.py +++ b/scripts/release_gate.py @@ -204,88 +204,179 @@ def check_security_zero_secrets() -> tuple[bool, str]: return True, "Zero secret files, live tokens, or obfuscated secret assignments in src/" -def check_production_update_feed() -> tuple[bool, str]: - """Live verification of public release feed manifest and package URL.""" +# ═══════════════════════════════════════════════════════════════ +# Publication Gate +# ═══════════════════════════════════════════════════════════════ +# +# Проверка публикации отделена от офлайновой части, потому что раньше они были +# смешаны и обе были беззубыми. Измерено на прежней реализации: +# - при полном обрыве сети возвращался PASS ("check skipped"); +# - при 404 на манифест возвращался PASS ("not yet published"); +# - при 404 на пакет возвращался PASS ("pending upload"); +# - при живом пакете печаталось PACKAGE_HASH_VERIFIED=True, хотя hashlib в +# файле не вызывался ни разу: скачивались байты 0-10 через заголовок Range, +# и этого хватало, чтобы объявить хеш проверенным. +# То есть ворота публикации пропускали релиз при любом исходе, включая полное +# отсутствие релиза. +# +# Теперь: офлайновые проверки (1-6) блокируют всегда; публикация проверяется +# по-настоящему — релиз есть, ассеты есть, пакет скачан целиком, SHA-256 +# сошёлся с опубликованным. Блокирует она в режиме публикации (--publication +# или HERMES_RELEASE_PUBLICATION_GATE=1); в обычном прогоне CI, где релиза для +# ветки нет и быть не должно, результат сообщается как есть и не блокирует. +# Неизмеренное называется "Н/Д" с причиной, а не выдаётся за проверенное. + +PUBLICATION_MODE_ENV = "HERMES_RELEASE_PUBLICATION_GATE" + +# Имена ассетов-установщиков; совпадают с выбором в update_manager. +PACKAGE_ASSET_NAMES = ("HermesHubSetup.exe", "hermes-hub-setup.sh", "install-linux.sh") +CHECKSUMS_ASSET_NAME = "checksums.txt" + +# Пакет качается целиком, поэтому размер ограничен: подставленный гигантский +# ассет не должен превращать ворота в отказ в обслуживании самим себе. +MAX_PACKAGE_BYTES = 512 * 1024 * 1024 + + +def is_publication_mode() -> bool: + """Требуется ли блокирующая проверка публикации.""" + return "--publication" in sys.argv or os.environ.get(PUBLICATION_MODE_ENV, "") == "1" + + +def _http_get(url: str, timeout: int = 30): import urllib.request - import urllib.error + req = urllib.request.Request( + url, headers={"User-Agent": f"HermesHub-ReleaseGate/{__version__}"} + ) + return urllib.request.urlopen(req, timeout=timeout) + + +def _download_and_hash(url: str) -> tuple[str, int]: + """Скачать поток целиком и посчитать SHA-256. Никаких частичных диапазонов.""" + import hashlib + digest = hashlib.sha256() + size = 0 + with _http_get(url, timeout=120) as resp: + while True: + chunk = resp.read(1024 * 256) + if not chunk: + break + size += len(chunk) + if size > MAX_PACKAGE_BYTES: + raise ValueError(f"пакет превышает {MAX_PACKAGE_BYTES} байт") + digest.update(chunk) + return digest.hexdigest(), size + + +def _parse_checksums(text: str) -> dict[str, str]: + """Разобрать строки вида ' <имя файла>'.""" + table: dict[str, str] = {} + for line in text.splitlines(): + parts = line.strip().split() + if len(parts) >= 2 and re.fullmatch(r"[0-9a-fA-F]{64}", parts[0]): + table[parts[-1].lstrip("*")] = parts[0].lower() + return table + + +def check_offline_update_contract() -> tuple[bool, str]: + """Офлайновая часть: адрес обновления входит в список разрешённых.""" from antigravity_provider.updater.update_manager import DEFAULT_UPDATE_URL, is_allowed_update_host if not is_allowed_update_host(DEFAULT_UPDATE_URL): - return False, f"Default update URL host not in allowlist: {DEFAULT_UPDATE_URL}" + return False, f"Адрес обновления вне списка разрешённых: {DEFAULT_UPDATE_URL}" + return True, f"Адрес обновления в списке разрешённых: {DEFAULT_UPDATE_URL}" + +def check_publication_gate() -> tuple[bool, str]: + """Релиз опубликован, ассеты на месте, SHA-256 пакета сошёлся. + + В режиме публикации любой недостижимый шаг — отказ. Вне его отказ не + блокирует релиз, но и не выдаётся за успех. + """ + import urllib.error + from antigravity_provider.updater.update_manager import DEFAULT_UPDATE_URL + + blocking = is_publication_mode() + + def verdict(ok: bool, msg: str) -> tuple[bool, str]: + if ok: + return True, msg + if blocking: + return False, msg + return True, f"[НЕ БЛОКИРУЕТ: режим публикации не запрошен] {msg}" + + # 1. Манифест релиза try: - req = urllib.request.Request( - DEFAULT_UPDATE_URL, - headers={"User-Agent": f"HermesHub-ReleaseGate/{__version__}"} - ) - with urllib.request.urlopen(req, timeout=6) as resp: - if resp.status == 200: - data = json.loads(resp.read().decode("utf-8-sig")) - p_ver = data.get("version") or data.get("tag_name", "").lstrip("v") - p_url = data.get("package_url") - if not p_url and data.get("assets"): - p_url = data["assets"][0].get("browser_download_url") - if not p_url: - p_url = data.get("html_url") or DEFAULT_UPDATE_URL - - if not p_ver: - return False, "Public update manifest is missing version or tag_name" - - # Verify package URL reachability - pkg_live = False - pkg_status = "UNKNOWN" - try: - head_req = urllib.request.Request( - p_url, - headers={"User-Agent": f"HermesHub-ReleaseGate/{__version__}"} - ) - # Use Range header to avoid downloading huge binaries - head_req.add_header("Range", "bytes=0-10") - with urllib.request.urlopen(head_req, timeout=6) as pkg_resp: - if pkg_resp.status in (200, 206, 302): - pkg_live = True - pkg_status = "PACKAGE_LIVE" - except urllib.error.HTTPError as pkg_he: - if pkg_he.code == 404: - pkg_status = "PENDING_RELEASE_UPLOAD_404" - else: - pkg_status = f"HTTP_{pkg_he.code}" - except Exception as pkg_ex: - pkg_status = f"CHECK_SKIPPED_{pkg_ex}" - - manifest_live = True - package_live = False - hash_verified = False - - if pkg_live: - package_live = True - # If package is live, verify hash on partial bytes or full stream - hash_verified = True - return True, f"[MANIFEST_LIVE=True, PACKAGE_LIVE=True, PACKAGE_HASH_VERIFIED=True] Manifest live (v{p_ver}) and release asset verified at {p_url}" - elif pkg_status == "PENDING_RELEASE_UPLOAD_404": - return True, ( - f"[MANIFEST_LIVE=True, PACKAGE_LIVE=False (Pending Upload 404), PACKAGE_HASH_VERIFIED=Offline Validated] " - f"Manifest is live (v{p_ver}), release zip ready for GitHub Release asset upload. Offline updater tests passed." - ) - else: - return True, ( - f"[MANIFEST_LIVE=True, PACKAGE_LIVE=False ({pkg_status}), PACKAGE_HASH_VERIFIED=Offline Validated] " - f"Manifest live (v{p_ver}). Offline updater tests passed." - ) - + with _http_get(DEFAULT_UPDATE_URL) as resp: + if resp.status != 200: + return verdict(False, f"Манифест релиза ответил HTTP {resp.status}") + data = json.loads(resp.read().decode("utf-8-sig")) except urllib.error.HTTPError as he: - if he.code == 404: - return True, f"[MANIFEST_LIVE=False, PACKAGE_LIVE=False] Public manifest not yet published (HTTP 404). Offline updater tests passed." - return False, f"HTTP Error checking update feed: {he}" + return verdict(False, f"Манифест релиза недоступен: HTTP {he.code} ({DEFAULT_UPDATE_URL})") except Exception as exc: - return True, f"[MANIFEST_LIVE=Unknown, PACKAGE_LIVE=Unknown] Public feed check skipped ({exc}). Offline updater tests passed." + return verdict(False, f"Манифест релиза недоступен: {type(exc).__name__}: {exc}") - return True, "Production update feed verified" + version = data.get("version") or str(data.get("tag_name", "")).lstrip("v") + if not version: + return verdict(False, "В манифесте релиза нет ни version, ни tag_name") + + # 2. Ассеты + assets: dict[str, str] = {} + for asset in data.get("assets") or []: + name = asset.get("name") + url = asset.get("browser_download_url") + if name and url: + assets[name] = url + if not assets and data.get("package_url"): + assets[Path(data["package_url"]).name] = data["package_url"] + + if not assets: + return verdict(False, f"У релиза v{version} нет ни одного ассета") + + packages = [n for n in PACKAGE_ASSET_NAMES if n in assets] + if not packages: + return verdict( + False, + f"У релиза v{version} нет ни одного пакета установки " + f"{PACKAGE_ASSET_NAMES}; опубликованы: {sorted(assets)}", + ) + + # 3. Опубликованные контрольные суммы + if CHECKSUMS_ASSET_NAME not in assets: + return verdict(False, f"У релиза v{version} нет {CHECKSUMS_ASSET_NAME}: сверять хеш не с чем") + try: + with _http_get(assets[CHECKSUMS_ASSET_NAME]) as resp: + published = _parse_checksums(resp.read().decode("utf-8", errors="replace")) + except Exception as exc: + return verdict(False, f"{CHECKSUMS_ASSET_NAME} не скачивается: {type(exc).__name__}: {exc}") + if not published: + return verdict(False, f"{CHECKSUMS_ASSET_NAME} не содержит ни одной строки с SHA-256") + + # 4. Полное скачивание и сверка хеша каждого пакета + verified = [] + for name in packages: + expected = published.get(name) + if not expected: + return verdict(False, f"Для {name} нет строки в {CHECKSUMS_ASSET_NAME}") + try: + actual, size = _download_and_hash(assets[name]) + except Exception as exc: + return verdict(False, f"{name} не скачивается целиком: {type(exc).__name__}: {exc}") + if actual != expected: + return verdict(False, f"SHA-256 {name} не сошёлся: опубликован {expected}, посчитан {actual}") + verified.append(f"{name} ({size} байт)") + + return True, ( + f"[RELEASE_LIVE=True, PACKAGES={len(verified)}, PACKAGE_HASH_VERIFIED=True] " + f"Релиз v{version}: пакеты скачаны целиком и сверены с {CHECKSUMS_ASSET_NAME} — " + + ", ".join(verified) + ) def run_release_gate(): print("=" * 70) print(f" Hermes Hub — Release Gate Verification Suite (Target: v{__version__})") + mode = "публикация (проверки 1-8 блокируют)" if is_publication_mode() else "офлайн (блокируют 1-7)" + print(f" Режим: {mode}") print("=" * 70) checks = [ @@ -295,7 +386,8 @@ def run_release_gate(): ("4. Full Offline Pytest Suite", "[INTEGRATION VERIFIED]", check_full_test_suite), ("5. Zero Hardcoded Developer Paths", "[STATIC VERIFIED]", check_zero_hardcoded_paths), ("6. Zero Credentials & AST Secret Scan", "[SECURITY VERIFIED]", check_security_zero_secrets), - ("7. Public Production Update Feed", "[LIVE STATUS]", check_production_update_feed), + ("7. Update Contract (offline)", "[STATIC VERIFIED]", check_offline_update_contract), + ("8. Publication Gate", "[LIVE VERIFIED]", check_publication_gate), ] all_passed = True diff --git a/src/antigravity_provider/router/telemetry_service.py b/src/antigravity_provider/router/telemetry_service.py index 05d7e21..a26b9d5 100644 --- a/src/antigravity_provider/router/telemetry_service.py +++ b/src/antigravity_provider/router/telemetry_service.py @@ -161,7 +161,11 @@ class TelemetryService: if p.is_file(): try: import yaml - data = yaml.safe_dump(p.read_text(encoding="utf-8")) + # safe_load, а не safe_dump: dump сериализует текст обратно + # в строку, isinstance(data, dict) никогда не выполнялось, и + # таблица цен из pricing.yaml не загружалась ни разу. Тихо: + # ошибку глушил except. + data = yaml.safe_load(p.read_text(encoding="utf-8")) if isinstance(data, dict) and "pricing" in data: self._pricing_table = dict(data["pricing"]) return diff --git a/src/antigravity_provider/router/web/server.py b/src/antigravity_provider/router/web/server.py index dda5fdc..0371e3b 100644 --- a/src/antigravity_provider/router/web/server.py +++ b/src/antigravity_provider/router/web/server.py @@ -7,6 +7,7 @@ import time import dataclasses import logging from typing import Any, Dict, List, Optional +from urllib.parse import urlparse from antigravity_provider import paths from antigravity_provider.version import __version__ @@ -105,6 +106,63 @@ def get_auth_token(x_hub_token: str = Header(None)) -> bool: raise HTTPException(status_code=401, detail="Invalid X-Hub-Token") return True +# Небезопасные методы принимаются только от собственного интерфейса. +# +# На loopback токен не требуется вовсе (см. get_auth_token), а POST /api/action +# меняет состояние: удаляет учётные данные, чистит аккаунты, переключает +# маршрутизацию, запускает входы OAuth. CORS от этого не защищает — он мешает +# ПРОЧИТАТЬ ответ, а не отправить запрос. +# +# Проверено на конфигурации по умолчанию (web_api_host=127.0.0.1): POST с +# Content-Type: text/plain уходит кросс-сайтом без предварительного запроса +# (простой запрос по правилам CORS), а request.json() разбирает тело +# независимо от Content-Type. Запрос с Origin: https://evil.example.com и без +# токена доходил до исполнителя действий — отвечало уже само действие. +# +# Проверяется Sec-Fetch-Site, а при его отсутствии — Origin против адреса, по +# которому пришёл запрос. Отсутствие обоих заголовков означает не-браузерного +# клиента (curl, собственный CLI): браузер на кросс-сайтовый POST заголовок +# Origin ставит обязательно, поэтому его отсутствие подделкой не является. +_SAME_SITE_FETCH_VALUES = {"same-origin", "none"} + + +def _configured_allowed_origins() -> set: + raw = str(_web_settings().get("web_api_allowed_origins", "")).split(",") + return {o.strip() for o in raw if o.strip() and o.strip() != "*"} + + +def require_same_origin(request: Request) -> bool: + """Отклонить межсайтовый небезопасный запрос до того, как он что-то изменит.""" + fetch_site = (request.headers.get("sec-fetch-site") or "").strip().lower() + if fetch_site: + if fetch_site in _SAME_SITE_FETCH_VALUES: + return True + if request.headers.get("origin", "") in _configured_allowed_origins(): + return True + raise HTTPException( + status_code=403, + detail=f"Межсайтовый запрос отклонён (Sec-Fetch-Site: {fetch_site})", + ) + + origin = request.headers.get("origin") + if not origin: + return True + + host = (request.headers.get("host") or "").strip().lower() + try: + origin_host = urlparse(origin).netloc.strip().lower() + except Exception: + origin_host = "" + if origin_host and host and origin_host == host: + return True + if origin in _configured_allowed_origins(): + return True + raise HTTPException( + status_code=403, + detail="Origin не совпадает с адресом хаба: межсайтовый запрос отклонён", + ) + + # Коммит и время запуска СНИМАЮТСЯ ОДИН РАЗ, при старте процесса. # # get_installed_commit() читает манифест с диска при каждом вызове, поэтому @@ -283,7 +341,11 @@ def get_snapshot(authorized: bool = Depends(get_auth_token)): return JSONResponse(content=jsonable_encoder(snap_dict)) @app.post("/api/action") -async def handle_action(request: Request, authorized: bool = Depends(get_auth_token)): +async def handle_action( + request: Request, + authorized: bool = Depends(get_auth_token), + same_origin: bool = Depends(require_same_origin), +): try: data = await request.json() except Exception: @@ -356,7 +418,11 @@ def get_skills_endpoint(authorized: bool = Depends(get_auth_token)): @app.post("/api/skills/assign") -async def assign_skill_endpoint(request: Request, authorized: bool = Depends(get_auth_token)): +async def assign_skill_endpoint( + request: Request, + authorized: bool = Depends(get_auth_token), + same_origin: bool = Depends(require_same_origin), +): """Assign a skill to a specific subagent.""" from antigravity_provider.router.skills_service import SkillsService try: @@ -377,7 +443,11 @@ async def assign_skill_endpoint(request: Request, authorized: bool = Depends(get @app.post("/api/skills/unassign") -async def unassign_skill_endpoint(request: Request, authorized: bool = Depends(get_auth_token)): +async def unassign_skill_endpoint( + request: Request, + authorized: bool = Depends(get_auth_token), + same_origin: bool = Depends(require_same_origin), +): """Remove an assigned skill from a subagent.""" from antigravity_provider.router.skills_service import SkillsService try: @@ -406,7 +476,11 @@ def get_skills_usage_endpoint(authorized: bool = Depends(get_auth_token)): @app.post("/api/skills/diagnose") -async def diagnose_skill_endpoint(request: Request, authorized: bool = Depends(get_auth_token)): +async def diagnose_skill_endpoint( + request: Request, + authorized: bool = Depends(get_auth_token), + same_origin: bool = Depends(require_same_origin), +): """Run SkillDoctor diagnostics on a skill by name, filepath, or raw content.""" from antigravity_provider.router.skills_service import SkillsService try: @@ -456,7 +530,11 @@ def get_compression_history_endpoint(limit: int = 20, authorized: bool = Depends @app.post("/api/compression/test") -async def test_compression_endpoint(request: Request, authorized: bool = Depends(get_auth_token)): +async def test_compression_endpoint( + request: Request, + authorized: bool = Depends(get_auth_token), + same_origin: bool = Depends(require_same_origin), +): """Execute test context compression on synthetic benchmark prompt.""" from antigravity_provider.router.settings_service import get_hub_settings from antigravity_provider.router.local_supervisor import LocalSupervisor diff --git a/tests/test_a37_isolation_guards.py b/tests/test_a37_isolation_guards.py index 8cc08f9..afb3e12 100644 --- a/tests/test_a37_isolation_guards.py +++ b/tests/test_a37_isolation_guards.py @@ -476,3 +476,70 @@ def test_normal_work_inside_project_still_allowed(guard_env): for cmd in ["rm src/temp_file.py", "rm -rf build/", "rm ./tests/tmp.log"]: allowed, reason, _alt = guard_env.validate_command(cmd) assert allowed, f"защита мешает штатной работе: {cmd} ({reason})" + + +# ── HUB-1: небезопасные методы не принимаются с чужой страницы ── + + +@pytest.fixture +def loopback_client(): + """Клиент при конфигурации по умолчанию: web_api_host=127.0.0.1, токен не нужен.""" + import antigravity_provider.router.web.server as srv + + with patch.object(srv, "_web_settings", return_value={"web_api_host": "127.0.0.1"}): + yield TestClient(app) + + +_UNSAFE_ENDPOINTS = [ + ("/api/action", '{"action": "clear_accounts", "data": {}}'), + ("/api/skills/assign", '{"skill": "x", "profile": "y"}'), + ("/api/skills/unassign", '{"skill": "x", "profile": "y"}'), + ("/api/skills/diagnose", '{"skill": "x"}'), + ("/api/compression/test", '{"text": "x"}'), +] + + +@pytest.mark.parametrize("path, body", _UNSAFE_ENDPOINTS) +def test_cross_site_post_is_rejected(loopback_client, path, body): + """Межсайтовый POST отклоняется до того, как что-либо изменит. + + На loopback токен не требуется, а действия меняют состояние: удаляют + учётные данные, чистят аккаунты, переключают маршрутизацию. CORS от этого + не защищает — он мешает прочитать ответ, а не отправить запрос. Измерено: + POST с Content-Type text/plain уходит кросс-сайтом без предварительного + запроса, request.json() разбирает тело независимо от Content-Type, и + запрос с чужим Origin доходил до исполнителя действий. + """ + headers = { + "Content-Type": "text/plain;charset=UTF-8", + "Origin": "https://evil.example.com", + "Sec-Fetch-Site": "cross-site", + } + res = loopback_client.post(path, content=body, headers=headers) + assert res.status_code == 403, f"{path} принял межсайтовый запрос: HTTP {res.status_code}" + + +@pytest.mark.parametrize("path, body", _UNSAFE_ENDPOINTS) +def test_cross_site_post_rejected_without_fetch_metadata(loopback_client, path, body): + """Браузер без Sec-Fetch-* всё равно ставит Origin — по нему и отклоняем.""" + headers = {"Content-Type": "text/plain", "Origin": "https://evil.example.com"} + res = loopback_client.post(path, content=body, headers=headers) + assert res.status_code == 403, f"{path} принял запрос с чужим Origin: HTTP {res.status_code}" + + +@pytest.mark.parametrize( + "label, headers", + [ + ("собственный интерфейс", {"Origin": "http://testserver", "Sec-Fetch-Site": "same-origin"}), + ("адресная строка", {"Sec-Fetch-Site": "none"}), + ("не-браузерный клиент", {}), + ], +) +def test_own_interface_and_cli_still_work(loopback_client, label, headers): + """Защита не должна мешать собственному интерфейсу и не-браузерным клиентам.""" + res = loopback_client.post( + "/api/action", + json={"action": "___нет_такого___", "data": {}}, + headers=headers, + ) + assert res.status_code != 403, f"{label} отклонён межсайтовой защитой" diff --git a/tests/test_a59_visible_update.py b/tests/test_a59_visible_update.py index 822ab09..bd08e7d 100644 --- a/tests/test_a59_visible_update.py +++ b/tests/test_a59_visible_update.py @@ -268,7 +268,10 @@ def test_p0_3_stop_running_hub_isolates_user_and_excludes_current_pid(simulated_ with patch("antigravity_provider.updater.update_manager.sys") as mock_sys: mock_sys.platform = "win32" if is_win else "linux" - with patch("subprocess.run") as mock_run: + # На Windows os.getuid не существует; ветка Linux падала бы на нём в + # общий except и возвращала «остановлено» никого не остановив. create=True + # позволяет подставить его там, где его нет. + with patch("os.getuid", return_value=1000, create=True), patch("subprocess.run") as mock_run: # wmic и pgrep перечисляют один и тот же набор: чужой PID и свой. mock_run.return_value = MagicMock(returncode=0, stdout=f"99999\n{current_pid}\n") diff --git a/tests/test_p0_release_gate.py b/tests/test_p0_release_gate.py index 5b1b2b0..eeef7cf 100644 --- a/tests/test_p0_release_gate.py +++ b/tests/test_p0_release_gate.py @@ -461,3 +461,163 @@ def test_s4_secret_scanner_ast_detection(tmp_path): clean_file.write_text('def hello(): return "world"\n', encoding="utf-8") v3 = scan_file_for_secrets(clean_file) assert len(v3) == 0 + + +# ── HUB-1: ворота публикации не пропускают релиз при любом исходе ── + + +def _load_release_gate(): + import importlib + import sys + scripts_dir = str(Path(__file__).resolve().parent.parent / "scripts") + if scripts_dir not in sys.path: + sys.path.insert(0, scripts_dir) + import release_gate + importlib.reload(release_gate) + return release_gate + + +@pytest.mark.unit +@pytest.mark.parametrize( + "failure, expected_in_message", + [ + ("network", "недоступен"), + ("http_404", "404"), + ("no_assets", "ассет"), + ("no_checksums", "checksums.txt"), + ("hash_mismatch", "не сошёлся"), + ], +) +def test_publication_gate_blocks_instead_of_failing_open(monkeypatch, failure, expected_in_message): + """Недостижимая публикация — отказ, а не PASS. + + Измерено на прежней реализации: при обрыве сети, при 404 на манифест и при + 404 на пакет возвращался PASS. Ворота пропускали релиз при любом исходе, + включая полное отсутствие релиза, а строка PACKAGE_HASH_VERIFIED=True + печаталась при том, что hashlib в файле не вызывался ни разу — хеш был + объявлен проверенным после чтения одиннадцати байт через заголовок Range. + """ + import urllib.error + release_gate = _load_release_gate() + monkeypatch.setenv(release_gate.PUBLICATION_MODE_ENV, "1") + monkeypatch.setattr(release_gate.sys, "argv", ["release_gate.py"]) + + manifest = { + "tag_name": "v9.9.9", + "assets": [ + {"name": "HermesHubSetup.exe", "browser_download_url": "https://example.invalid/setup.exe"}, + {"name": "checksums.txt", "browser_download_url": "https://example.invalid/checksums.txt"}, + ], + } + if failure == "no_assets": + manifest["assets"] = [] + if failure == "no_checksums": + manifest["assets"] = [manifest["assets"][0]] + + class _Resp: + status = 200 + + def __init__(self, payload: bytes): + self._payload = payload + + def read(self, *_a): + payload, self._payload = self._payload, b"" + return payload + + def __enter__(self): + return self + + def __exit__(self, *_a): + return False + + def fake_get(url, timeout=30): + if failure == "network": + raise urllib.error.URLError("сети нет") + if failure == "http_404": + raise urllib.error.HTTPError(url, 404, "Not Found", {}, None) + if url.endswith("checksums.txt"): + return _Resp(b"%s HermesHubSetup.exe\n" % (b"a" * 64)) + if url.endswith("setup.exe"): + return _Resp(b"payload-with-a-different-hash") + return _Resp(json.dumps(manifest).encode("utf-8")) + + monkeypatch.setattr(release_gate, "_http_get", fake_get) + + ok, msg = release_gate.check_publication_gate() + assert ok is False, f"ворота пропустили релиз при отказе '{failure}': {msg}" + assert expected_in_message in msg, f"причина отказа не названа: {msg!r}" + + +@pytest.mark.unit +def test_publication_gate_hashes_the_whole_package(monkeypatch): + """Успех объявляется только после полного скачивания и сверки SHA-256.""" + import hashlib + release_gate = _load_release_gate() + monkeypatch.setenv(release_gate.PUBLICATION_MODE_ENV, "1") + monkeypatch.setattr(release_gate.sys, "argv", ["release_gate.py"]) + + package = b"hermes hub installer payload" + real_sha = hashlib.sha256(package).hexdigest() + read_bytes = {"total": 0} + + class _Resp: + status = 200 + + def __init__(self, payload: bytes, count: bool = False): + self._payload = payload + self._count = count + + def read(self, *_a): + payload, self._payload = self._payload, b"" + if self._count: + read_bytes["total"] += len(payload) + return payload + + def __enter__(self): + return self + + def __exit__(self, *_a): + return False + + manifest = { + "tag_name": "v9.9.9", + "assets": [ + {"name": "HermesHubSetup.exe", "browser_download_url": "https://example.invalid/setup.exe"}, + {"name": "checksums.txt", "browser_download_url": "https://example.invalid/checksums.txt"}, + ], + } + + def fake_get(url, timeout=30): + if url.endswith("checksums.txt"): + return _Resp(f"{real_sha} HermesHubSetup.exe\n".encode("utf-8")) + if url.endswith("setup.exe"): + return _Resp(package, count=True) + return _Resp(json.dumps(manifest).encode("utf-8")) + + monkeypatch.setattr(release_gate, "_http_get", fake_get) + + ok, msg = release_gate.check_publication_gate() + assert ok is True, msg + assert "PACKAGE_HASH_VERIFIED=True" in msg + assert read_bytes["total"] == len(package), ( + f"пакет должен быть прочитан целиком, прочитано {read_bytes['total']} из {len(package)}" + ) + + +@pytest.mark.unit +def test_offline_run_does_not_claim_publication_verified(monkeypatch): + """Без режима публикации отсутствие релиза не блокирует, но и не врёт.""" + import urllib.error + release_gate = _load_release_gate() + monkeypatch.delenv(release_gate.PUBLICATION_MODE_ENV, raising=False) + monkeypatch.setattr(release_gate.sys, "argv", ["release_gate.py"]) + + def fake_get(url, timeout=30): + raise urllib.error.HTTPError(url, 404, "Not Found", {}, None) + + monkeypatch.setattr(release_gate, "_http_get", fake_get) + + ok, msg = release_gate.check_publication_gate() + assert ok is True, "обычный прогон CI не должен блокироваться отсутствием релиза" + assert "НЕ БЛОКИРУЕТ" in msg + assert "PACKAGE_HASH_VERIFIED=True" not in msg, "непроверенное не должно объявляться проверенным"