HUB-1: зелёный main — граница workspace и UTF-8 в verification #2
5 changed files with 284 additions and 22 deletions
|
|
@ -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,
|
||||
|
|
|
|||
48
src/antigravity_provider/console_encoding.py
Normal file
48
src/antigravity_provider/console_encoding.py
Normal 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)
|
||||
|
|
@ -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:
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -396,3 +396,83 @@ 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})"
|
||||
|
|
|
|||
Loading…
Reference in a new issue