HUB-1: зелёный main — граница workspace и UTF-8 в verification #2

Merged
ochenstarik-ui merged 10 commits from hub/audit-p0-green-main into main 2026-09-03 17:59:50 +00:00
16 changed files with 1394 additions and 172 deletions

View file

@ -6,11 +6,22 @@ on:
pull_request:
branches: [ main ]
# Матрица из двух систем.
#
# Обе джобы стояли на windows-latest, и это дорого обошлось: инвариант A37 не
# держался на Windows, а тесты установки и остановки процессов молча
# предполагали Linux. Прогон на одной системе не показывал ни того, ни другого.
# Проект работает на Linux и активно получает Linux-правки, поэтому обе системы
# проверяются одинаковым набором.
jobs:
test:
name: Clean Windows Runner Test
runs-on: windows-latest
name: Clean Runner Test (${{ matrix.os }})
runs-on: ${{ matrix.os }}
timeout-minutes: 15
strategy:
fail-fast: false
matrix:
os: [windows-latest, ubuntu-latest]
steps:
- name: Checkout repository
@ -40,9 +51,13 @@ jobs:
python scripts/release_gate.py
headless:
name: Headless Run (no GUI dependencies)
runs-on: windows-latest
name: Headless Run (${{ matrix.os }}, no GUI dependencies)
runs-on: ${{ matrix.os }}
timeout-minutes: 15
strategy:
fail-fast: false
matrix:
os: [windows-latest, ubuntu-latest]
steps:
- name: Checkout repository
uses: actions/checkout@v4

View file

@ -57,6 +57,21 @@ jobs:
$manifest | ConvertTo-Json -Depth 5 | Out-File -FilePath "$distDir/update_manifest.json" -Encoding utf8
Write-Host "Generated update_manifest.json with SHA256: $hash"
# Набор проверяется ДО публикации.
#
# update_manager ищет в релизе строго HermesHubSetup.exe или
# hermes-hub-setup.sh, а шаг выше собирает только zip и манифест. Такой
# релиз становится "latest", и любая попытка обновиться отвечает «в
# релизе не найден подходящий файл обновления для текущей платформы».
#
# Раньше это не проявлялось лишь потому, что весь конвейер падал на шаге
# Release Gate — на тех же двух дефектах, что и CI; ни один его прогон не
# доходил до публикации, а релизы выкладывались мимо него. Как только
# тесты позеленели, случайная защита исчезла.
- name: Built assets must be installable by the updater
run: |
python scripts/release_gate.py --assets dist
- name: Publish GitHub Release
uses: softprops/action-gh-release@v2
with:
@ -68,3 +83,12 @@ jobs:
prerelease: false
env:
GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
# Ворота публикации: проверяют опубликованный релиз, а не сборку.
# Релиз есть, ассеты есть, пакет скачан целиком, SHA-256 сошёлся с
# опубликованным checksums.txt. Здесь они блокируют: раньше эта проверка
# возвращала PASS при обрыве сети, при 404 на манифест и при 404 на
# пакет, то есть пропускала релиз при любом исходе.
- name: Publication Gate (published release must be verifiable)
run: |
python scripts/release_gate.py --publication-only

View file

@ -0,0 +1,248 @@
# Отчёт HUB-1: зелёный main и P0 из аудита
## Сдача
| | |
|---|---|
| Ветка | `hub/audit-p0-green-main` |
| `START_HEAD` | `93da1b22fd2e0385f46b71a1e220fa1e4e716545` |
| Последний рабочий коммит | `f061961600dd2eb3d6901cc154dbf5a82456dd14` |
| `origin/main` на момент сдачи | `93da1b2` (не двигался) |
| PR | https://github.com/ochenstarik-ui/hermes-hub/pull/2 |
| Зелёный прогон CI | https://github.com/ochenstarik-ui/hermes-hub/actions/runs/33729660525 |
| `git status` | чисто (вне репозитория лежит посторонний `gyoza_shorts.mp4`, не мой и не тронут) |
### Зелёный CI — все четыре джоба
| Джоб | Итог |
|---|---|
| Clean Runner Test (windows-latest) | **pass** |
| Clean Runner Test (ubuntu-latest) | **pass** |
| Headless Run (windows-latest) | **pass** |
| Headless Run (ubuntu-latest) | **pass** |
`ruff check .` — чисто. Release Gate — PASSED на обеих системах.
Локально (Linux): **778 passed, 2 skipped, 4 deselected**. База до работы —
739 passed, 2 skipped. Число тестов выросло на 39, ни один не удалён.
---
## P0-1. Зелёный main
### Обе причины из задания подтвердились — и обе оказались шире описания
**1. Инвариант A37 не держался на Windows.** Причина именно та, что
предполагалась. Воспроизведено локально на окружении Windows (нет переменной
`HOME`): `os.path.expandvars("$HOME/.hermes")` оставляет строку как есть, путь
перестаёт быть абсолютным, склеивается с каталогом проекта и оказывается
«внутри разрешённого корня» — команда проходит.
Заодно нашлась **зеркальная дыра, в задании не названная**: на Linux так же
проходили `rm -rf %USERPROFILE%\.hermes` и `del /f /q C:\Windows\System32`.
`Path("C:/Windows").resolve()` на Linux приписывает пути текущий каталог, и
удаление системного каталога Windows выглядело работой внутри проекта.
Разбор пути сведён в один конвейер, как и требовало задание: классификация
диалекта shell **по самой команде, а не по системе-хозяину** → раскрытие
распознанных переменных, с разрешением `HOME`/`USERPROFILE` в домашний каталог
даже когда их нет в окружении → нормализация разделителей → канонизация →
сравнение с защищёнными корнями. Каждый несостоявшийся шаг **закрывает
проход**: непроверяемый путь не считается разрешённым. Через тот же конвейер
пропущены `validate_path`, `is_forbidden_path`, `is_inside_allowed_root`.
Доказательство: новые тесты воспроизводят окружение обеих систем на любой из
них. На прежнем guard они падают — **6 failed**, ровно на дефекте из CI и на
зеркальных случаях; на новом проходят. `test_a37_isolation_guards` зелёный на
Windows-раннере.
**2. UTF-8 ронял verification-скрипт.** Воспроизведено точно: строка 63, тот же
`UnicodeEncodeError`. Общий помощник `console_encoding.force_utf8_output`
ставит UTF-8 на потоки и оставляет запасной путь, если поток перекодировать
нельзя. Той же реализацией заменён самодельный блок в `cli_commands`.
Скрипт проходит **10/10** под `PYTHONIOENCODING=cp1252` и под `ascii`.
### Причин красного CI было не две, а семь
Это главное расхождение с заданием. Ревьюер видел две; живой прогон на
Windows после их устранения показал ещё пять. Четыре из них — **не дефекты
продукта, а допущения тестов, зашитые под Linux**:
1. `test_a41` читал вывод скрипта в кодировке системы. Скрипт стал писать
UTF-8, а родитель на Windows читал трубу как cp1252 и разваливался на
`UnicodeDecodeError`, оставляя `proc.stdout` равным `None`. Кодировка
задана явно с обеих сторон трубы.
2. `test_p0_3_stop_running_hub` знал только про ветку Linux (`os.kill` по
списку от `pgrep`). На Windows процессы останавливает `taskkill` по списку
от `wmic`. Инвариант один — «чужой процесс останавливается, свой PID не
трогаем» — теперь проверяется на обеих ветках.
3-4. Оба теста установки подсовывали bash-скрипт `hermes-hub-setup.sh`; на
Windows выбирается `HermesHubSetup.exe`, и установка честно отвечала «в
релизе не найден подходящий файл обновления». Установщик берётся под ту
систему, на которой идёт прогон. Проверка сообщения смотрит, назван ли код
возврата, а не на склонение: ветки формулируют «код 3» и «кодом 3».
Пятая — моя собственная: добавленный русский вывод Release Gate уронил шаг
на cp1252. Тот же класс дефекта, то же лекарство.
Шестая — **флейк, из-за которого main краснел случайно**: базовый прогон до
начала работы дал то 738, то 739. Причина найдена по падению ubuntu-джоба:
`test_seq_token_prevents_stale_refresh_clobber` сравнивал поколения до и после
устаревшего вызова, а `HubStateStore` — процессный синглтон, и фоновый сборщик
квот от другого теста успевает поднять `generation` между вызовами. Теперь
проверяется инвариант (устаревший ответ отброшен ровно один раз, состояние
назад не откатывается), а не равенство. Пять полных прогонов со случайным
порядком — 777 passed.
---
## P0-2. Остальные P0 аудита — каждый подтверждён исполнением
### 1. Release Gate заявлял проверку хеша, которой не было — **подтвердилось**
Хуже, чем в аудите. `hashlib` в `scripts/release_gate.py` **не вызывался ни
разу**: скачивались байты 0-10 через заголовок `Range`, и этого хватало, чтобы
напечатать `PACKAGE_HASH_VERIFIED=True`. «Проверенным ассетом» при этом
оказывался первый в списке — `checksums.txt`, а не пакет.
### 2. Publication gate fail-open — **подтвердилось**, во всех трёх условиях
Измерено прогоном самой функции:
| Условие | Прежний вердикт |
|---|---|
| Полный обрыв сети | **PASS** |
| Манифест 404 (релиза нет) | **PASS** |
| Пакет 404 (ассет не загружен) | **PASS** |
Ворота пропускали релиз при любом исходе, включая полное отсутствие релиза.
Разделено, как требовало задание: офлайновая часть (версии, тесты, updater,
статика, секреты, список разрешённых адресов) блокирует всегда; Publication
Gate проверяет, что релиз есть, ассеты есть, пакет скачан **целиком** и
SHA-256 сошёлся с опубликованным `checksums.txt`. Блокирует в режиме
публикации (`--publication` или `HERMES_RELEASE_PUBLICATION_GATE=1`); в
обычном прогоне CI, где релиза для ветки нет и быть не должно, результат
сообщается как есть и не блокирует. Неизмеренное называется причиной, а не
выдаётся за проверенное.
Проверено на живом релизе `v0.1.3-b1`: два пакета скачаны целиком, хеши
сошлись. Проверено на отказах: обрыв сети и 404 теперь **FAIL**.
### 3. localhost `/api/action` без CSRF — **подтвердилось**
CORS уже закрыт правкой ревьюера, но CORS мешает **прочитать** ответ, а не
**отправить** запрос. Измерено на конфигурации по умолчанию
(`web_api_host=127.0.0.1`): POST с `Content-Type: text/plain` уходит
кросс-сайтом без предварительного запроса (простой запрос по правилам CORS), а
`request.json()` разбирает тело независимо от `Content-Type`. Запрос с
`Origin: https://evil.example.com` и без токена доходил до исполнителя
действий — отвечало уже само действие. Среди доступных действий
`clear_accounts`, `delete_credentials`, `set_main`.
Проверяется `Sec-Fetch-Site`, при его отсутствии — `Origin` против адреса
запроса. Замер после правки:
| Запрос | Итог |
|---|---|
| чужой сайт, `Sec-Fetch-Site: cross-site` | **403** |
| чужой сайт, старый браузер (только `Origin`) | **403** |
| собственный интерфейс | 200 |
| адресная строка / расширение | 200 |
| не-браузерный клиент (curl, CLI) | 200 |
Защита распространена на все пять небезопасных методов, не только на
`/api/action`.
---
## P1
1. **Zip-slip — НЕ ВОСПРОИЗВОДИТСЯ.** Это единственное расхождение с аудитом
по существу, и оно в пользу продукта. Архив с `../`, с абсолютным путём и с
записью-ссылкой распакован через `zipfile.extractall`: ничего за пределы
каталога не вышло, абсолютный путь стал относительным, `../` схлопнулись, а
запись-ссылка легла обычным файлом. CPython санирует пути сам.
Но это свойство реализации, а не обещание формата, и распаковка идёт в
корень установки. Граница сделана собственным инвариантом: каждая запись
проверяется до записи на диск, отклоняются абсолютные пути, выход через
`..`, ссылки и записи не-файлового типа. Инвариант закреплён тестом, а не
оставлен на усмотрение стандартной библиотеки.
2. **pricing fallback** — исправлено: `safe_load` вместо `safe_dump`. `dump`
сериализовал текст обратно в строку, `isinstance(data, dict)` не
выполнялось никогда, таблица цен не загружалась ни разу, а `except` это
глушил.
3. **CI-матрица Windows + Linux** — сделано, обе джобы. Именно отсутствие
Linux-джоба и позволяло четырём платформенным допущениям прятаться; на
первом же прогоне матрицы Linux-джоб поймал флейк, который Windows не
показывал.
4. Прочее из аудита (failover error policy, `uv sync --frozen`, лишний `web`
extra, secret-scan шире) — не трогал, по заданию это отдельные задания.
---
## Ограничения задания — соблюдены
- Правки ревьюера из `main` не откатывались; в `main` напрямую не пушил.
- Фронтенд не трогал: npm, сборки и фреймворков не добавлено.
- Проверка SHA-256 **усилена**, а не ослаблена; список разрешённых адресов не
тронут.
- Учётные данные и `~/.hermes/agy_profiles/` не тронуты.
- Версия `0.1.3` не понижена.
- Неизмеренное названо причиной: Publication Gate вне режима публикации
печатает «НЕ БЛОКИРУЕТ» с причиной и не заявляет `PACKAGE_HASH_VERIFIED`.
---
## Найдено сверх задания: релизный конвейер был мёртв, и мой же фикс снимал с него защиту
Это самое важное из того, что не значилось ни в задании, ни в аудите.
**Каждый прогон `Release Pipeline` завершался ошибкой** — все пять последних,
включая тег текущего релиза `v0.1.3-b1`. Причина ровно та же, что у красного
CI: шаг `Run Release Gate Check` падал на `test_a37_isolation_guards` и
`test_a41_clean_install`. До публикации не доходил ни один прогон, а релизы
выкладывались мимо конвейера.
**Отсюда ловушка.** `release.yml` собирает `hermes-hub-<версия>.zip` и
`update_manifest.json`, а `update_manager` ищет в релизе строго
`HermesHubSetup.exe` или `hermes-hub-setup.sh`. Настоящие релизы содержат
`HermesHubSetup.exe`, `hermes-hub-setup.sh` и `checksums.txt` — то есть
собраны не этим конвейером. Пока тесты были красными, конвейер падал и ничего
не публиковал; **как только я их починил, случайная защита исчезла**: первый
же тег привёл бы к публикации «latest» без установщиков, и любая попытка
обновиться отвечала бы «В релизе не найден подходящий файл обновления для
текущей платформы».
Ловушка закрыта явно, до публикации: шаг `Built assets must be installable by
the updater` (`release_gate.py --assets dist`) проверяет, что собранный набор
содержит установщик и `checksums.txt`, и падает с названной причиной и
подсказкой про `installer/build_installer.ps1` и
`installer/build_installer_linux.sh`. После публикации добавлен шаг
`Publication Gate` (`release_gate.py --publication-only`) — тот самый строгий
режим, ради которого ворота и разделялись.
Чего я **не** делал: не переписывал сборку установщиков в `release.yml`.
Проверить это можно только выкладыванием настоящего релиза по тегу, а это
решение владельца, не исполнителя. Конвейер по-прежнему не доходит до
публикации — но теперь падает с честной причиной вместо чужой.
## Что стоит решить ревьюеру
- Джобы переименованы (`Clean Windows Runner Test` → `Clean Runner Test
(windows-latest)`). Защиты ветки на `main` сейчас нет, так что ничего не
сломалось; если её будут включать — имена проверок брать новые.
- Publication Gate по умолчанию не блокирует. Это осознанный выбор: иначе
каждый PR краснел бы за отсутствие релиза для ветки. В `release.yml` он уже
встроен и блокирует (`--publication-only`, после публикации). Вручную:
`python scripts/release_gate.py --publication`.
- **Главное решение — сборка установщиков в `release.yml`.** Конвейер собирает
zip, которым обновиться нельзя. Скрипты `installer/build_installer.ps1` и
`installer/build_installer_linux.sh` в репозитории есть, но Linux-установщик
требует Linux-раннера, то есть релизной джобе нужна матрица. Работа
небольшая, но проверяется только настоящей публикацией по тегу — поэтому
оставлена за владельцем.

View file

@ -23,9 +23,14 @@ ROOT = Path(__file__).resolve().parent.parent
if str(ROOT / "src") not in sys.path:
sys.path.insert(0, str(ROOT / "src"))
from antigravity_provider.console_encoding import force_utf8_output
from antigravity_provider.version import __version__, get_version
from antigravity_provider import paths
# Отчёт ворот печатается по-русски, а консоль Windows-раннера — cp1252.
# Ставится до первого вывода: иначе падает вывод, а не проверки.
force_utf8_output()
def check_version_consistency() -> tuple[bool, str]:
ver = get_version()
@ -204,88 +209,218 @@ 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]:
"""Разобрать строки вида '<sha256> <имя файла>'."""
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 check_publishable_assets(dist_dir: Path) -> tuple[bool, str]:
"""Собранный набор ассетов действительно устанавливается обновлением.
Проверяется до публикации. Причина: update_manager ищет в релизе строго
HermesHubSetup.exe или hermes-hub-setup.sh/install-linux.sh, а release.yml
собирает hermes-hub-<версия>.zip и update_manifest.json. Такой релиз
становится "latest", и на любой попытке обновиться владелец получает
"В релизе не найден подходящий файл обновления для текущей платформы".
Раньше это не проявлялось лишь потому, что весь релизный конвейер падал
на тех же двух дефектах, что и CI: каждый его прогон завершался ошибкой, а
релизы публиковались мимо него. Как только тесты позеленели, случайная
защита исчезла поэтому набор проверяется явно.
"""
if not dist_dir.is_dir():
return False, f"Каталог сборки не найден: {dist_dir}"
present = {item.name for item in dist_dir.iterdir() if item.is_file()}
installers = sorted(present & set(PACKAGE_ASSET_NAMES))
problems = []
if not installers:
problems.append(
f"нет ни одного установщика {list(PACKAGE_ASSET_NAMES)}"
f"обновление такой релиз поставить не сможет"
)
if CHECKSUMS_ASSET_NAME not in present:
problems.append(f"нет {CHECKSUMS_ASSET_NAME} — сверять хеш пакета будет не с чем")
if problems:
return False, (
f"Набор ассетов в {dist_dir} непригоден для публикации: "
+ "; ".join(problems)
+ f". Собрано: {sorted(present)}. Установщики собираются скриптами "
f"installer/build_installer.ps1 и installer/build_installer_linux.sh"
)
return True, f"Набор ассетов пригоден для публикации: {installers} + {CHECKSUMS_ASSET_NAME}"
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 +430,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
@ -319,5 +455,24 @@ def run_release_gate():
sys.exit(1)
def _run_single(title: str, check) -> None:
"""Выполнить одну проверку и завершиться её итогом."""
print("=" * 70)
print(f" Hermes Hub — {title}")
print("=" * 70)
ok, msg = check()
print(f" {'[OK]' if ok else '[FAIL]'} {msg}")
sys.exit(0 if ok else 1)
if __name__ == "__main__":
run_release_gate()
if "--assets" in sys.argv:
index = sys.argv.index("--assets")
target = Path(sys.argv[index + 1]) if len(sys.argv) > index + 1 else ROOT / "dist"
_run_single("Publishable Assets Check", lambda: check_publishable_assets(target))
elif "--publication-only" in sys.argv:
# Запускается ПОСЛЕ публикации: проверяет опубликованный релиз, а не сборку.
os.environ[PUBLICATION_MODE_ENV] = "1"
_run_single("Publication Gate", check_publication_gate)
else:
run_release_gate()

View file

@ -18,6 +18,13 @@ for p in [
if p.is_dir() and str(p) not in sys.path:
sys.path.insert(0, str(p))
from antigravity_provider.console_encoding import force_utf8_output
# Отчёт печатается по-русски, а консоль Windows-раннера в CI — cp1252: без этого
# первый же [PASS] с кириллицей роняет скрипт UnicodeEncodeError'ом ещё до того,
# как проверки что-либо покажут. Ставится до первого вывода.
force_utf8_output()
from antigravity_provider.router.router_config import (
RolePolicy,
RouterConfig,

View file

@ -0,0 +1,48 @@
"""Принудительный UTF-8 для потоков вывода.
Инструменты Hermes печатают по-русски, а консоль Windows-раннера в CI работает
в cp1252. Первый же `print` с кириллицей роняет процесс UnicodeEncodeError'ом —
падает вывод, не логика. Измерено на `scripts/verify_multi_provider_router.py`:
строка `[PASS] Чистая конфигурация...` обрывала прогон с кодом 1.
Модуль ставит UTF-8 на stdout/stderr и оставляет запасной путь на случай, когда
перекодировать поток нельзя: тогда непечатаемые символы заменяются, но процесс
продолжает работу. Вывод инструмента не должен быть причиной падения.
"""
from __future__ import annotations
import sys
from typing import Any, Iterable
__all__ = ["force_utf8_output"]
def _reconfigure(stream: Any) -> bool:
"""Перевести один поток на UTF-8. True, если получилось."""
reconfigure = getattr(stream, "reconfigure", None)
if reconfigure is None:
return False
for errors in ("strict", "backslashreplace"):
try:
reconfigure(encoding="utf-8", errors=errors)
return True
except Exception:
continue
# Поток не перекодировать (подменён, закрыт, не текстовый). Тогда хотя бы
# снимем строгость с текущей кодировки, чтобы кириллица не роняла процесс.
try:
reconfigure(errors="backslashreplace")
return True
except Exception:
return False
def force_utf8_output(streams: Iterable[str] = ("stdout", "stderr")) -> None:
"""Перевести стандартные потоки на UTF-8; молча пропустить недоступные.
Вызывается один раз на старте точки входа, до первого вывода.
"""
for name in streams:
stream = getattr(sys, name, None)
if stream is not None:
_reconfigure(stream)

View file

@ -9,17 +9,16 @@ import time
from pathlib import Path
from typing import Any, Dict, List, Optional
from antigravity_provider.console_encoding import force_utf8_output
from antigravity_provider.router.router_config import RouterConfig, RouterProfileConfig, load_router_config
from antigravity_provider.router.health_tracker import HealthTracker
from antigravity_provider.router.router_engine import RouterEngine, get_router_engine
from antigravity_provider.router.adapters import get_adapter
from antigravity_provider.router.profile_manager import ProfileAuthManager, mask_email, mask_id
if hasattr(sys.stdout, "reconfigure"):
try:
sys.stdout.reconfigure(encoding="utf-8")
except Exception:
pass
# Вывод CLI — по-русски, консоль Windows по умолчанию не UTF-8. Общий помощник,
# тот же, что и у скриптов проверки: одна реализация, одно поведение.
force_utf8_output()
def print_router_status() -> int:

View file

@ -135,6 +135,117 @@ DESTRUCTIVE_COMMAND_NAMES: Set[str] = {
}
# ── Единый конвейер разбора пути ────────────────────────────────
#
# Граница вокруг агентских shell-действий обязана работать одинаково на всех
# поддерживаемых системах, иначе доказанной она не является ни на одной.
# Измерено: "rm -rf $HOME/.hermes" отклонялось на Linux и проходило на Windows,
# потому что переменной HOME в окружении Windows нет — os.path.expandvars
# оставлял "$HOME" как есть, путь переставал быть абсолютным, склеивался с
# каталогом проекта и оказывался "внутри разрешённого корня". Зеркальная дыра
# на Linux: "%USERPROFILE%\.hermes" и "C:\Windows" тоже проходили.
#
# Порядок шагов: классификация диалекта → раскрытие распознанных переменных →
# нормализация разделителей → канонизация → сравнение с корнями. Каждый шаг,
# который не удался, закрывает проход: непроверяемый путь не считается
# разрешённым.
_HOME_VARIABLE_NAMES = frozenset({"HOME", "USERPROFILE"})
_VARIABLE_REFERENCE = re.compile(
r"\$\{(?P<brace>[A-Za-z_][A-Za-z0-9_]*)\}"
r"|\$(?P<bare>[A-Za-z_][A-Za-z0-9_]*)"
r"|%(?P<percent>[A-Za-z_][A-Za-z0-9_]*)%"
)
_WINDOWS_DRIVE_PREFIX = re.compile(r"^[A-Za-z]:[\\/]")
# Признаки записи в диалекте cmd/PowerShell: %VAR%, буква диска, разделитель "\".
_WINDOWS_DIALECT_MARKERS = re.compile(r"%[A-Za-z_][A-Za-z0-9_]*%|[A-Za-z]:[\\/]|\\[A-Za-z0-9_.]")
def looks_like_windows_dialect(cmd_line: str) -> bool:
"""Записана ли команда в диалекте Windows.
Диалект определяется по самой строке, а не по системе-хозяину: команду в
записи cmd могут прислать и на Linux, и разобрать её posix-правилами нельзя
shlex съест "\" как экранирование и разделитель пути исчезнет.
"""
return bool(_WINDOWS_DIALECT_MARKERS.search(cmd_line))
def expand_path_argument(raw: str) -> Tuple[str, Optional[str]]:
"""Раскрыть "~" и переменные окружения обоих диалектов.
Возвращает (раскрытая строка, причина нераскрытия | None). Имена HOME и
USERPROFILE разрешаются в домашний каталог даже тогда, когда их нет в
окружении: команда, написанная в чужом диалекте, целит ровно туда же.
Всё, что раскрыть не удалось, возвращается причиной вызывающий обязан
закрыться, а не гадать.
"""
unresolved: List[str] = []
def _substitute(match: "re.Match[str]") -> str:
name = match.group("brace") or match.group("bare") or match.group("percent")
value = os.environ.get(name)
if value is None and name.upper() in _HOME_VARIABLE_NAMES:
try:
value = str(Path.home())
except Exception:
value = None
if value is None:
unresolved.append(name)
return match.group(0)
return value
expanded = _VARIABLE_REFERENCE.sub(_substitute, raw)
expanded = os.path.expanduser(expanded)
if expanded.startswith("~"):
# expanduser не смог определить домашний каталог: оставлять "~" внутри
# пути нельзя — он перестанет быть абсолютным и уедет внутрь проекта.
unresolved.append("~")
if unresolved:
return expanded, "не раскрыты: " + ", ".join(sorted(set(unresolved)))
return expanded, None
def canonical_path(raw: Path | str, base_cwd: Optional[Path | str] = None) -> Tuple[Optional[Path], Optional[str]]:
"""Привести аргумент к каноническому пути или объяснить отказ.
Ровно одно из двух возвращаемых значений не None. Отказ это отказ в
доступе: путь, который нельзя достоверно разрешить, нельзя и признать
находящимся внутри разрешённого корня.
"""
text = str(raw)
if not text:
return None, "пустой путь"
expanded, unresolved_reason = expand_path_argument(text)
if unresolved_reason:
return None, f"путь '{raw}' невозможно раскрыть ({unresolved_reason})"
normalized = expanded.replace("\\", "/")
if _WINDOWS_DRIVE_PREFIX.match(normalized) and not Path(normalized).is_absolute():
# Путь с буквой диска на не-Windows: канонизировать его нечем. resolve()
# припишет ему текущий каталог, и "C:/Windows" окажется внутри проекта.
return None, f"путь '{raw}' записан в диалекте другой системы и здесь не проверяем"
candidate = Path(normalized)
if not candidate.is_absolute() and not normalized.startswith("/"):
try:
base = Path(base_cwd) if base_cwd is not None else paths.get_repo_root()
except Exception as exc:
return None, f"не определён базовый каталог для '{raw}': {exc}"
candidate = base / normalized
try:
return candidate.resolve(), None
except Exception as exc:
return None, f"путь '{raw}' не разрешается: {exc}"
class WorkspaceBoundaryGuard:
"""Enforces explicit workspace boundaries, defends credential directories, and inspects destructive operations."""
@ -206,7 +317,11 @@ class WorkspaceBoundaryGuard:
def is_inside_allowed_root(self, path: Path | str) -> bool:
"""Check whether the given path resolves within any allowed root."""
try:
target = Path(path).expanduser().resolve()
target, resolve_error = canonical_path(path)
if target is None:
# Путь не разрешается — считать его находящимся внутри
# разрешённого корня нельзя.
return False
for root in self.get_allowed_roots():
try:
target.relative_to(root)
@ -220,7 +335,9 @@ class WorkspaceBoundaryGuard:
def is_forbidden_path(self, path: Path | str) -> Tuple[bool, Optional[str]]:
"""Check whether the path touches an unconditionally protected directory or file."""
try:
target = Path(path).expanduser().resolve()
target, resolve_error = canonical_path(path)
if target is None:
return True, resolve_error
# 1. Exact match or child of forbidden directory
for fpath in self.get_forbidden_paths():
if target == fpath:
@ -252,10 +369,9 @@ class WorkspaceBoundaryGuard:
Returns: (is_allowed: bool, reason: str, safe_alternative: Optional[str])
"""
try:
target = Path(path).expanduser().resolve()
except Exception as exc:
return False, f"Недопустимый путь '{path}': {exc}", "Используйте стандартный относительный путь"
target, resolve_error = canonical_path(path)
if target is None:
return False, f"Недопустимый путь: {resolve_error}", "Укажите путь внутри проекта явно, без нераскрытых переменных"
# Check unconditional forbidden paths for mutating/deleting operations
if operation in {"delete", "write", "truncate", "move"}:
@ -290,12 +406,18 @@ class WorkspaceBoundaryGuard:
# Parse command tokens
if isinstance(cmd_line, list):
tokens = list(cmd_line)
windows_dialect = any(looks_like_windows_dialect(str(t)) for t in tokens)
else:
# Диалект берётся из самой команды: строку в записи cmd нельзя
# разбирать posix-правилами — shlex съест "\\" как экранирование,
# и разделитель пути исчезнет ещё до проверки.
windows_dialect = looks_like_windows_dialect(cmd_line)
try:
# Windows and POSIX-compatible shlex split
tokens = shlex.split(cmd_line, posix=(os.name != "nt"))
tokens = shlex.split(cmd_line, posix=not windows_dialect)
except Exception:
tokens = cmd_line.split()
if windows_dialect:
tokens = [t[1:-1] if len(t) > 1 and t[0] == t[-1] and t[0] in "\"'" else t for t in tokens]
if not tokens:
return True, "OK", None
@ -310,7 +432,7 @@ class WorkspaceBoundaryGuard:
# Extract target arguments (skip flags starting with - or /)
targets = []
for arg in tokens[1:]:
if arg.startswith("-") or (os.name == "nt" and arg.startswith("/") and len(arg) == 2):
if arg.startswith("-") or (windows_dialect and arg.startswith("/") and len(arg) == 2):
continue
targets.append(arg)
@ -321,17 +443,23 @@ class WorkspaceBoundaryGuard:
return False, f"Команда {cmd_name} запущена в недопустимом каталоге: {reason}", alt
else:
for target_arg in targets:
# Тильда и переменные окружения раскрываются ДО проверки.
# Тильда и переменные окружения раскрываются ДО проверки, и
# одинаково для обоих диалектов.
#
# Без этого "rm -rf ~/.hermes/agy_profiles" не считался
# абсолютным путём, склеивался с каталогом проекта в путь с
# буквальным "~" внутри и признавался допустимым. Проверено:
# команда с тильдой проходила, та же команда с абсолютным
# путём отклонялась. То есть самый естественный способ
# написать опасную команду обходил защиту ровно там, ради
# чего она и делалась — на каталоге учётных данных.
expanded = os.path.expandvars(os.path.expanduser(target_arg))
target_path = Path(expanded) if Path(expanded).is_absolute() else (base_cwd / expanded)
# буквальным "~" внутри и признавался допустимым; а
# "rm -rf $HOME/.hermes" ровно так же проходил на Windows,
# где переменной HOME в окружении нет. Самый естественный
# способ написать опасную команду обходил защиту ровно там,
# ради чего она и делалась — на каталоге учётных данных.
target_path, resolve_error = canonical_path(target_arg, base_cwd=base_cwd)
if target_path is None:
return (
False,
f"Команда '{cmd_name}' обращается к непроверяемому пути: {resolve_error}",
"Укажите путь внутри проекта явно, без нераскрытых переменных",
)
ok, reason, alt = self.validate_path(target_path, operation="delete")
if not ok:
return False, f"Команда '{cmd_name}' пытается удалить недопустимый путь '{target_arg}': {reason}", alt

View file

@ -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

View file

@ -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

View file

@ -19,6 +19,7 @@ import logging
import os
import re
import shutil
import zipfile
import subprocess
import sys
import threading
@ -351,6 +352,43 @@ def acknowledge_last_applied_update() -> None:
logger.debug("Failed acknowledging last_applied_update: %s", exc)
def _extract_within(zf: "zipfile.ZipFile", dest: Path) -> None:
"""Распаковать архив, не выпуская ни одной записи за пределы dest.
Отклоняются: абсолютные пути, выход через "..", символические ссылки и
любые записи, не являющиеся обычным файлом или каталогом.
"""
import stat as _stat
root = dest.resolve()
for info in zf.infolist():
# Биты типа файла проставлены не всегда: архиватор мог записать только
# права доступа. Судим лишь тогда, когда тип действительно указан, —
# иначе обычный файл с правами 0o600 выглядел бы записью чужого типа.
file_type = _stat.S_IFMT(info.external_attr >> 16)
if file_type and file_type not in (_stat.S_IFREG, _stat.S_IFDIR):
raise ValueError(
f"Пакет обновления содержит запись недопустимого типа: {info.filename!r}"
)
name = info.filename.replace("\\", "/")
if name.startswith("/") or re.match(r"^[A-Za-z]:", name):
raise ValueError(f"Пакет обновления содержит абсолютный путь: {info.filename!r}")
target = (root / name).resolve()
if target != root and root not in target.parents:
raise ValueError(
f"Пакет обновления пытается записать за пределы каталога установки: {info.filename!r}"
)
if info.is_dir():
target.mkdir(parents=True, exist_ok=True)
continue
target.parent.mkdir(parents=True, exist_ok=True)
with zf.open(info, "r") as source, open(target, "wb") as out:
shutil.copyfileobj(source, out)
def stop_running_hub(timeout_sec: float = 10.0) -> bool:
"""Останавливает только процессы хаба текущего пользователя, исключая текущий PID."""
current_pid = os.getpid()
@ -1212,8 +1250,6 @@ class UpdateManager:
def apply_update_sync(self, package_zip: Path, target_dir: Optional[Path] = None) -> Tuple[bool, str]:
"""Apply update package with automatic backup and rollback on failure."""
import zipfile
dest = target_dir or paths.get_repo_root()
backup = self.backup_dir
backup.mkdir(parents=True, exist_ok=True)
@ -1229,8 +1265,15 @@ class UpdateManager:
shutil.copytree(src_item, dst_item)
# 2. Extract update package into dest
#
# Каждая запись проверяется до записи на диск. Измерено, что
# extractall в CPython уже отбрасывает "..", ведущие разделители и
# буквы дисков, а запись-ссылку кладёт обычным файлом: побега из
# каталога добиться не удалось. Но это свойство реализации, а не
# обещание формата — а распаковка идёт в корень установки. Проверка
# делает границу собственным инвариантом, который виден в тестах.
with zipfile.ZipFile(package_zip, "r") as zf:
zf.extractall(dest)
_extract_within(zf, dest)
# 3. Verify syntax and integrity of updated python files in src/
import py_compile

View file

@ -396,3 +396,150 @@ def test_destructive_command_with_tilde_is_rejected(monkeypatch, tmp_path):
# Обычная работа внутри проекта не должна страдать.
allowed, _reason, _alt = guard.validate_command("rm src/temp_file.py")
assert allowed, "защита мешает штатной работе внутри проекта"
# ── HUB-1: граница обязана держаться одинаково на Windows и Linux ──
#
# Инвариант A37 падал только на Windows-раннере: "rm -rf $HOME/.hermes"
# проходил мимо защиты, потому что переменной HOME в окружении Windows нет.
# Прогон на одной системе этого не показывал. Тесты ниже воспроизводят
# окружение обеих систем на любой из них, поэтому дыра больше не может
# спрятаться за тем, где именно запущен CI.
_DESTRUCTIVE_BOTH_DIALECTS = [
"rm -rf ~/.hermes/agy_profiles",
"rm -rf ~/.ssh",
"rm -rf $HOME/.hermes",
"rm -rf ${HOME}/.hermes",
r"rm -rf %USERPROFILE%\.hermes",
r"del /f /q %USERPROFILE%\.ssh",
]
@pytest.fixture
def guard_env(monkeypatch, tmp_path):
"""Guard с HERMES_HOME во временном каталоге — вне домашнего и вне проекта."""
from antigravity_provider.router.security_guard import WorkspaceBoundaryGuard
monkeypatch.setenv("HERMES_HOME", str(tmp_path / "hermes"))
return WorkspaceBoundaryGuard()
@pytest.mark.parametrize("simulated_os", ["linux", "windows"])
@pytest.mark.parametrize("cmd", _DESTRUCTIVE_BOTH_DIALECTS)
def test_home_directed_destruction_rejected_on_both_systems(monkeypatch, guard_env, simulated_os, cmd):
"""Удаление по домашнему каталогу отклоняется в обоих окружениях.
Окружение Windows отличается от Linux ровно тем, из-за чего защита и
расходилась: HOME не задан, домашний каталог известен через USERPROFILE.
"""
if simulated_os == "windows":
monkeypatch.delenv("HOME", raising=False)
monkeypatch.setenv("USERPROFILE", str(Path.home()))
else:
monkeypatch.setenv("HOME", str(Path.home()))
monkeypatch.delenv("USERPROFILE", raising=False)
allowed, reason, _alt = guard_env.validate_command(cmd)
assert not allowed, f"[{simulated_os}] команда прошла мимо защиты: {cmd} ({reason})"
@pytest.mark.parametrize("simulated_os", ["linux", "windows"])
def test_guard_fails_closed_on_unresolvable_argument(monkeypatch, guard_env, simulated_os):
"""Нераскрытая переменная — отказ, а не пропуск.
Раньше "$HOME/.hermes" с неизвестной переменной переставал быть абсолютным
путём, склеивался с каталогом проекта и признавался допустимым. Путь,
который нельзя разрешить, нельзя и признать безопасным.
"""
if simulated_os == "windows":
monkeypatch.delenv("HOME", raising=False)
monkeypatch.delenv("HERMES_UNSET_TARGET", raising=False)
for cmd in ["rm -rf $HERMES_UNSET_TARGET/data", r"rm -rf %HERMES_UNSET_TARGET%\data"]:
allowed, reason, _alt = guard_env.validate_command(cmd)
assert not allowed, f"[{simulated_os}] guard не закрылся на нераскрытом пути: {cmd} ({reason})"
def test_foreign_dialect_absolute_path_is_not_treated_as_project_local(guard_env):
"""Путь с буквой диска на Linux не должен считаться внутренним.
Path("C:/Windows").resolve() на Linux приписывает пути текущий каталог, и
удаление системного каталога Windows выглядело как работа внутри проекта.
"""
allowed, reason, _alt = guard_env.validate_command(r"del /f /q C:\Windows\System32")
assert not allowed, f"путь чужого диалекта признан внутренним: {reason}"
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} отклонён межсайтовой защитой"

View file

@ -249,7 +249,14 @@ def test_p0_5_verification_script_runs_clean_and_filled(clean_a41_env):
env["HERMES_HOME"] = str(hermes_home)
script_path = Path(__file__).resolve().parent.parent / "scripts" / "verify_multi_provider_router.py"
proc = subprocess.run([sys.executable, str(script_path)], env=env, capture_output=True, text=True)
# Кодировка задаётся явно с обеих сторон трубы. Скрипт печатает по-русски и
# сам переводит вывод на UTF-8; родитель же на Windows по умолчанию читает
# трубу в кодировке системы (cp1252) и разваливается на UnicodeDecodeError,
# оставляя proc.stdout равным None — падал разбор вывода, а не проверки.
run_kwargs = dict(env=env, capture_output=True, text=True, encoding="utf-8", errors="replace")
proc = subprocess.run([sys.executable, str(script_path)], **run_kwargs)
assert proc.returncode == 0, f"Verification failed on clean config: {proc.stderr}\n{proc.stdout}"
assert "10/10 CHECKS PASSED" in proc.stdout
@ -259,6 +266,6 @@ def test_p0_5_verification_script_runs_clean_and_filled(clean_a41_env):
AutoAssigner.assign_profile_to_role("codex-1", "manager", is_primary=True)
AutoAssigner.assign_profile_to_role("ag-1", "manager", is_primary=False)
proc_filled = subprocess.run([sys.executable, str(script_path)], env=env, capture_output=True, text=True)
proc_filled = subprocess.run([sys.executable, str(script_path)], **run_kwargs)
assert proc_filled.returncode == 0, f"Verification failed on filled config: {proc_filled.stderr}\n{proc_filled.stdout}"
assert "10/10 CHECKS PASSED" in proc_filled.stdout

View file

@ -9,9 +9,11 @@ Verifies:
"""
from __future__ import annotations
import contextlib
import io
import json
import os
import re
import shutil
import sys
import tempfile
@ -249,23 +251,44 @@ def test_p0_2_sha256_mismatch_aborts_and_sets_failed_status(tmp_path, monkeypatc
# ── TEST 5: P0-3 Process Isolation stop_running_hub ──
@pytest.mark.parametrize("simulated_os", ["linux", "windows"])
@pytest.mark.unit
def test_p0_3_stop_running_hub_isolates_user_and_excludes_current_pid():
"""stop_running_hub filters by current UID on Linux and never targets own PID."""
def test_p0_3_stop_running_hub_isolates_user_and_excludes_current_pid(simulated_os):
"""Чужие процессы хаба останавливаются, собственный — никогда.
Проверяется на обеих ветках, а не на той, где случился прогон. Ветки
останавливают процессы по-разному: на Linux os.kill по списку от pgrep,
на Windows taskkill по списку от wmic. Тест знал только про первую и на
Windows-раннере падал на пустом списке убитых, хотя проверять надо один и
тот же инвариант «свой PID не трогаем».
"""
current_pid = os.getpid()
is_win = simulated_os == "windows"
# Mock subprocess.run for pgrep
with patch("subprocess.run") as mock_run:
# Simulate pgrep returning other PID and own PID
mock_run.return_value = MagicMock(returncode=0, stdout=f"99999 {current_pid}\n")
with patch("antigravity_provider.updater.update_manager.sys") as mock_sys:
mock_sys.platform = "win32" if is_win else "linux"
with patch("os.kill") as mock_kill:
stop_running_hub(timeout_sec=0.1)
# На 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")
# Check that kill was called on 99999 but NEVER on current_pid
killed_pids = [call.args[0] for call in mock_kill.call_args_list]
assert 99999 in killed_pids
assert current_pid not in killed_pids, "stop_running_hub must never kill current PID"
with patch("os.kill") as mock_kill:
stop_running_hub(timeout_sec=0.1)
if is_win:
killed_pids = [
int(call.args[0][-1])
for call in mock_run.call_args_list
if call.args and call.args[0] and call.args[0][0] == "taskkill"
]
else:
killed_pids = [call.args[0] for call in mock_kill.call_args_list]
assert 99999 in killed_pids, f"[{simulated_os}] чужой процесс хаба не остановлен: {killed_pids}"
assert current_pid not in killed_pids, f"[{simulated_os}] остановлен собственный процесс"
# ── TEST 6: P0-3 apply_update_sync Rollback on Corruption ──
@ -381,40 +404,43 @@ def test_p0_4_failed_install_records_nothing_and_names_exit_code(tmp_path, monke
monkeypatch.setenv("HERMES_HOME", str(tmp_path / "hermes"))
monkeypatch.setenv("HERMES_HUB_DEV_MODE", "1")
installer = tmp_path / "hermes-hub-setup.sh"
installer.write_bytes(b"#!/bin/bash\nexit 3\n")
sha = hashlib.sha256(installer.read_bytes()).hexdigest()
with platform_installer(tmp_path, exit_code=3) as (asset_name, installer):
sha = hashlib.sha256(installer.read_bytes()).hexdigest()
check_result = UpdateCheckResult(
update_available=True,
current_version="0.1.3",
latest_version="0.1.4",
latest_commit="deadbeefdeadbeef",
installed_commit="0000000aaaa",
assets={
"hermes-hub-setup.sh": f"file://{installer}",
"checksums.txt": "file:///nonexistent",
},
)
check_result = UpdateCheckResult(
update_available=True,
current_version="0.1.3",
latest_version="0.1.4",
latest_commit="deadbeefdeadbeef",
installed_commit="0000000aaaa",
assets={
asset_name: f"file://{installer}",
"checksums.txt": "file:///nonexistent",
},
)
mgr = UpdateManager()
real_download = UpdateManager._download_file
mgr = UpdateManager()
real_download = UpdateManager._download_file
def fake_download(self, url, dest, progress_cb=None):
if dest.name == "checksums.txt":
dest.write_text(f"{sha} hermes-hub-setup.sh\n", encoding="utf-8")
return
return real_download(self, url, dest, progress_cb)
def fake_download(self, url, dest, progress_cb=None):
if dest.name == "checksums.txt":
dest.write_text(f"{sha} {asset_name}\n", encoding="utf-8")
return
return real_download(self, url, dest, progress_cb)
with patch.object(UpdateManager, "_download_file", fake_download):
with patch("antigravity_provider.updater.update_manager.stop_running_hub", return_value=True):
ok, msg = mgr.install_latest_update(check_result=check_result)
with patch.object(UpdateManager, "_download_file", fake_download):
with patch("antigravity_provider.updater.update_manager.stop_running_hub", return_value=True):
ok, msg = mgr.install_latest_update(check_result=check_result)
assert ok is False
assert get_last_applied_update() is None, (
"После провалившейся установки записи о применённом обновлении быть не должно"
)
assert "код 3" in msg, f"Причина отказа должна называть код возврата, получено: {msg!r}"
# Проверяется, что код назван, а не как он склоняется: ветки формулируют
# по-разному («код 3» и «кодом 3»), инвариант же один.
assert re.search(r"код\w*\s+3", msg), (
f"Причина отказа должна называть код возврата, получено: {msg!r}"
)
prog = UpdateManager.get_progress_dict()
assert prog["status"] == "failed"
@ -422,6 +448,31 @@ def test_p0_4_failed_install_records_nothing_and_names_exit_code(tmp_path, monke
UpdateManager._set_progress(status="idle", message="Готов к обновлению")
# ── Установщик под ту систему, на которой идёт прогон ──
#
# Ветки установки различаются: на Windows выбирается HermesHubSetup.exe и
# запускается через Popen, на Linux — hermes-hub-setup.sh через bash. Тесты
# ниже проверяют не установщик, а учёт его результата, поэтому подставляется
# тот файл, который данная система действительно выбирает. Раньше в них был
# зашит bash-скрипт, и на Windows-раннере установка отвечала «в релизе не
# найден подходящий файл обновления» — падало допущение теста, не продукт.
@contextlib.contextmanager
def platform_installer(tmp_path, exit_code: int):
"""Отдать (имя ассета, путь) и заставить установщик вернуть exit_code."""
if sys.platform == "win32":
installer = tmp_path / "HermesHubSetup.exe"
# Содержимое не исполняется: запуск подменён, проверяется учёт кода.
installer.write_bytes(b"MZ\x90\x00 hermes hub test installer\n")
with patch("subprocess.Popen") as mock_popen:
mock_popen.return_value = MagicMock(wait=MagicMock(return_value=exit_code))
yield "HermesHubSetup.exe", installer
else:
installer = tmp_path / "hermes-hub-setup.sh"
installer.write_bytes(f"#!/bin/bash\nexit {exit_code}\n".encode("utf-8"))
yield "hermes-hub-setup.sh", installer
# ── TEST 10: P0-4 успешная установка запись всё-таки делает ──
@pytest.mark.unit
def test_p0_4_successful_install_records_previous_and_new_build(tmp_path, monkeypatch):
@ -431,35 +482,34 @@ def test_p0_4_successful_install_records_previous_and_new_build(tmp_path, monkey
monkeypatch.setenv("HERMES_HOME", str(tmp_path / "hermes"))
monkeypatch.setenv("HERMES_HUB_DEV_MODE", "1")
installer = tmp_path / "hermes-hub-setup.sh"
installer.write_bytes(b"#!/bin/bash\nexit 0\n")
sha = hashlib.sha256(installer.read_bytes()).hexdigest()
with platform_installer(tmp_path, exit_code=0) as (asset_name, installer):
sha = hashlib.sha256(installer.read_bytes()).hexdigest()
check_result = UpdateCheckResult(
update_available=True,
current_version="0.1.3",
latest_version="0.1.4",
latest_commit="deadbeefdeadbeef",
installed_commit="0000000aaaa",
assets={
"hermes-hub-setup.sh": f"file://{installer}",
"checksums.txt": "file:///nonexistent",
},
)
check_result = UpdateCheckResult(
update_available=True,
current_version="0.1.3",
latest_version="0.1.4",
latest_commit="deadbeefdeadbeef",
installed_commit="0000000aaaa",
assets={
asset_name: f"file://{installer}",
"checksums.txt": "file:///nonexistent",
},
)
mgr = UpdateManager()
real_download = UpdateManager._download_file
mgr = UpdateManager()
real_download = UpdateManager._download_file
def fake_download(self, url, dest, progress_cb=None):
if dest.name == "checksums.txt":
dest.write_text(f"{sha} hermes-hub-setup.sh\n", encoding="utf-8")
return
return real_download(self, url, dest, progress_cb)
def fake_download(self, url, dest, progress_cb=None):
if dest.name == "checksums.txt":
dest.write_text(f"{sha} {asset_name}\n", encoding="utf-8")
return
return real_download(self, url, dest, progress_cb)
with patch.object(UpdateManager, "_download_file", fake_download):
with patch("antigravity_provider.updater.update_manager.stop_running_hub", return_value=True):
with patch.object(UpdateManager, "schedule_restart", return_value=(True, "перезапуск запущен")):
ok, msg = mgr.install_latest_update(check_result=check_result)
with patch.object(UpdateManager, "_download_file", fake_download):
with patch("antigravity_provider.updater.update_manager.stop_running_hub", return_value=True):
with patch.object(UpdateManager, "schedule_restart", return_value=(True, "перезапуск запущен")):
ok, msg = mgr.install_latest_update(check_result=check_result)
assert ok is True
rec = get_last_applied_update()
@ -506,3 +556,59 @@ def test_p0_2_app_js_handles_refused_cancel():
assert "setInterval(pollUpdateProgress" in tail, (
"после отклонённой отмены опрос хода загрузки должен возобновляться"
)
# ── HUB-1: распаковка обновления не выпускает записи за пределы каталога ──
@pytest.mark.unit
def test_update_package_cannot_write_outside_target(tmp_path):
"""Ни одна запись архива не должна оказаться вне каталога установки.
Измерено на CPython: extractall сам отбрасывает "..", ведущие разделители
и буквы дисков, а запись-ссылку кладёт обычным файлом побега добиться не
удалось, вопреки формулировке аудита. Но это свойство реализации, а не
обещание формата, и распаковка идёт в корень установки. Тест закрепляет
границу как собственный инвариант.
"""
import stat as _stat
import zipfile as _zipfile
from antigravity_provider.updater.update_manager import _extract_within
dest = tmp_path / "dest"
dest.mkdir()
outside = tmp_path / "outside.txt"
hostile = [
("выход через ..", "../outside.txt"),
("абсолютный путь", "/etc/passwd"),
("путь с буквой диска", "C:/Windows/x.txt"),
]
for index, (label, arcname) in enumerate(hostile):
archive = tmp_path / f"hostile_{index}.zip"
with _zipfile.ZipFile(archive, "w") as zf:
zf.writestr(arcname, "побег")
with _zipfile.ZipFile(archive) as zf:
with pytest.raises(ValueError):
_extract_within(zf, dest)
assert not outside.exists(), f"{label}: запись оказалась вне каталога установки"
# Символическая ссылка — тоже отказ, а не молчаливая распаковка файлом.
link_zip = tmp_path / "link.zip"
with _zipfile.ZipFile(link_zip, "w") as zf:
info = _zipfile.ZipInfo("link")
info.external_attr = (_stat.S_IFLNK | 0o777) << 16
zf.writestr(info, "/etc/passwd")
with _zipfile.ZipFile(link_zip) as zf:
with pytest.raises(ValueError):
_extract_within(zf, dest)
# Обычный пакет распаковывается как прежде.
good = tmp_path / "good.zip"
with _zipfile.ZipFile(good, "w") as zf:
zf.writestr("src/module.py", "x = 1\n")
zf.writestr("assets/logo.txt", "logo")
with _zipfile.ZipFile(good) as zf:
_extract_within(zf, dest)
assert (dest / "src" / "module.py").read_text(encoding="utf-8") == "x = 1\n"
assert (dest / "assets" / "logo.txt").is_file()

View file

@ -461,3 +461,204 @@ 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, "непроверенное не должно объявляться проверенным"
@pytest.mark.unit
def test_publishable_assets_check_rejects_uninstallable_release(tmp_path):
"""Набор без установщика не должен уходить в публикацию.
update_manager ищет в релизе HermesHubSetup.exe или hermes-hub-setup.sh, а
release.yml собирает только zip и манифест. Такой релиз становится
«latest», и обновление отвечает «в релизе не найден подходящий файл
обновления для текущей платформы». Раньше это не проявлялось лишь потому,
что весь релизный конвейер падал на шаге Release Gate на тех же двух
дефектах, что и CI, и до публикации не доходил ни один его прогон.
"""
release_gate = _load_release_gate()
as_built_today = tmp_path / "dist_zip_only"
as_built_today.mkdir()
(as_built_today / "hermes-hub-0.1.3.zip").write_bytes(b"zip")
(as_built_today / "update_manifest.json").write_text("{}", encoding="utf-8")
ok, msg = release_gate.check_publishable_assets(as_built_today)
assert ok is False, "набор без установщика признан пригодным к публикации"
assert "HermesHubSetup.exe" in msg
assert "checksums.txt" in msg
without_checksums = tmp_path / "dist_no_sums"
without_checksums.mkdir()
(without_checksums / "HermesHubSetup.exe").write_bytes(b"exe")
ok, msg = release_gate.check_publishable_assets(without_checksums)
assert ok is False, "набор без checksums.txt признан пригодным"
assert "checksums.txt" in msg
as_published_really = tmp_path / "dist_full"
as_published_really.mkdir()
for name in ("HermesHubSetup.exe", "hermes-hub-setup.sh", "checksums.txt"):
(as_published_really / name).write_bytes(b"x")
ok, msg = release_gate.check_publishable_assets(as_published_really)
assert ok is True, msg
ok, msg = release_gate.check_publishable_assets(tmp_path / "нет-такого")
assert ok is False, "отсутствующий каталог сборки должен быть отказом"

View file

@ -72,15 +72,27 @@ def test_seq_token_prevents_stale_refresh_clobber():
seq_fresh = store.next_seq()
snap_fresh = store.refresh(force_scan=False, seq=seq_fresh)
gen_fresh = snap_fresh.generation
skipped_before = store.refresh_skipped_total
# Simulate a delayed/stale response from an earlier seq counter
seq_stale = seq_fresh - 1
snap_after_stale = store.refresh(force_scan=False, seq=seq_stale)
# Stale response must be rejected, retaining the fresh generation
assert snap_after_stale.generation == gen_fresh
# Устаревший ответ должен быть отброшен.
#
# Проверяется именно отбрасывание, а не равенство поколений. HubStateStore —
# процессный синглтон, и фоновый сборщик квот, оставшийся от другого теста,
# успевает поднять generation между двумя вызовами. Прежнее
# `generation == gen_fresh` падало на этом с «assert 32 == 31» — примерно раз
# на десяток прогонов, только при случайном порядке тестов. Инвариант же
# другой: устаревший ответ отбрасывается, состояние назад не откатывается.
assert store.refresh_skipped_total == skipped_before + 1, (
"устаревший ответ должен быть отброшен ровно один раз"
)
assert snap_after_stale.generation >= gen_fresh, (
"состояние откатилось назад: устаревший ответ затёр более свежее"
)
assert snap_after_stale.seq != seq_stale
assert store.refresh_skipped_total >= 1
@pytest.mark.unit