diff --git a/src/antigravity_provider/router/action_handler.py b/src/antigravity_provider/router/action_handler.py index 5878d02..7dbde56 100644 --- a/src/antigravity_provider/router/action_handler.py +++ b/src/antigravity_provider/router/action_handler.py @@ -102,7 +102,7 @@ def do_test_profile(provider: str, profile_id: str) -> Dict[str, Any]: EventLogService.get().log('system', f'Сбой проверки {profile_id} ({model}): {e}', level='error') return {'success': False, 'model': model, 'duration_sec': round(time.time() - t0, 2), 'error': str(e)} -def do_delete_credentials(provider: str, profile_id: str) -> Tuple[bool, str]: +def do_delete_credentials(provider: str, profile_id: str, actor: str = "system") -> Tuple[bool, str]: # Сигнатура get_profile_dir — (profile_id, provider), а здесь её звали # наоборот. Внутри есть костыль, молча исправляющий перестановку, но только # для antigravity, openai-codex и opencode-go. Для grok, claude и local путь @@ -115,9 +115,26 @@ def do_delete_credentials(provider: str, profile_id: str) -> Tuple[bool, str]: if auth_p.is_file(): try: auth_p.unlink() - EventLogService.get().log('account', f'Учетные данные для {profile_id} удалены.', level='warning') + EventLogService.get().log( + 'account', + f'Учетные данные для {profile_id} удалены.', + level='warning', + actor=actor, + action='delete_credentials', + target_profile=profile_id, + outcome='success', + ) return True, f"Учетные данные для '{profile_id}' удалены" except Exception as e: + EventLogService.get().log( + 'account', + f'Ошибка удаления учётных данных {profile_id}: {e}', + level='error', + actor=actor, + action='delete_credentials', + target_profile=profile_id, + outcome='failed', + ) return False, f'Ошибка удаления: {e}' # Отсутствие файла — это не успех удаления. Раньше такой ответ выглядел # для пользователя как «сработало», хотя аккаунт оставался подключённым. @@ -285,7 +302,7 @@ class ActionExecutor: """Shared execution layer for Desktop and Web actions.""" @classmethod - def execute(cls, action: str, data: Dict[str, Any], async_runner: Optional[Callable] = None) -> Dict[str, Any]: + def execute(cls, action: str, data: Dict[str, Any], async_runner: Optional[Callable] = None, actor: str = "user:web") -> Dict[str, Any]: """ Execute the specified action. If async_runner is provided, long actions will be dispatched to it. @@ -544,8 +561,61 @@ class ActionExecutor: return {'ok': res.get('success', False), 'message': res.get('response') or res.get('error'), 'data': res} elif action == 'delete_credentials': - ok, msg = do_delete_credentials(prov, pid) + dry_run = bool(data.get('dry_run', False)) + confirmed = bool(data.get('confirmed', True)) + from antigravity_provider.router.profile_manager import get_profile_auth_path + auth_p = get_profile_auth_path(prov, pid) + if dry_run: + exists = auth_p.is_file() + EventLogService.get().log( + 'security', + f'Сухой прогон удаления учётных данных {pid}', + actor=actor, + action='delete_credentials', + target_profile=pid, + outcome='dry_run', + level='info', + ) + return { + 'ok': True, + 'dry_run': True, + 'message': f"Сухой прогон: будет удалён файл {auth_p.name} ({'найден' if exists else 'не найден'})", + 'data': {'path': str(auth_p), 'exists': exists, 'provider': prov, 'profile_id': pid}, + } + if not confirmed: + EventLogService.get().log( + 'security', + f'Запрошено подтверждение удаления учётных данных {pid}', + actor=actor, + action='delete_credentials', + target_profile=pid, + outcome='denied', + level='warning', + ) + return { + 'ok': False, + 'confirmation_required': True, + 'message': f"Требуется подтверждение удаления учётных данных для '{pid}'", + 'data': {'path': str(auth_p), 'provider': prov, 'profile_id': pid}, + } + + ok, msg = do_delete_credentials(prov, pid, actor=actor) return {'ok': ok, 'message': msg} + + elif action == 'dry_run_delete': + from antigravity_provider.router.security_guard import get_workspace_guard + targets = data.get('paths') or ([data.get('path')] if data.get('path') else []) + guard = get_workspace_guard() + res = guard.dry_run_deletion(targets) + EventLogService.get().log( + 'security', + f"Сухой прогон удаления: {res['total_files']} файлов, {res['total_dirs']} каталогов (риск: {res['risk_level']})", + actor=actor, + action='dry_run_delete', + outcome='dry_run', + level='info', + ) + return {'ok': True, 'message': 'Сухой прогон выполнен', 'data': res} elif action == 'auto_assign_all': if async_runner: diff --git a/src/antigravity_provider/router/security_guard.py b/src/antigravity_provider/router/security_guard.py new file mode 100644 index 0000000..8bfb901 --- /dev/null +++ b/src/antigravity_provider/router/security_guard.py @@ -0,0 +1,563 @@ +"""Security perimeter, boundary enforcement, credential protection, and forensic guards for Hermes Hub.""" +from __future__ import annotations + +import os +import re +import shlex +import shutil +import tempfile +import urllib.parse +from pathlib import Path +from typing import Any, Dict, List, Optional, Set, Tuple + +from antigravity_provider import paths + +# ═══════════════════════════════════════════════════════════════ +# Exceptions +# ═══════════════════════════════════════════════════════════════ + + +class SecurityViolationError(Exception): + """Base exception for security perimeter violations.""" + + def __init__(self, message: str, safe_alternative: Optional[str] = None, violation_type: str = "general"): + super().__init__(message) + self.message = message + self.safe_alternative = safe_alternative + self.violation_type = violation_type + + +class BoundaryViolationError(SecurityViolationError): + """Raised when an operation attempts to access or mutate paths outside allowed workspace.""" + + def __init__(self, message: str, safe_alternative: Optional[str] = None): + super().__init__(message, safe_alternative=safe_alternative, violation_type="boundary") + + +class CredentialProtectionError(SecurityViolationError): + """Raised when an operation attempts to directly delete or corrupt protected credential files.""" + + def __init__(self, message: str, safe_alternative: Optional[str] = None): + super().__init__(message, safe_alternative=safe_alternative, violation_type="credentials") + + +class NetworkBoundaryViolationError(SecurityViolationError): + """Raised when an outbound network request targets a host not in the allowed destination whitelist.""" + + def __init__(self, message: str, safe_alternative: Optional[str] = None): + super().__init__(message, safe_alternative=safe_alternative, violation_type="network") + + +# ═══════════════════════════════════════════════════════════════ +# Secret Scrubbing Utilities +# ═══════════════════════════════════════════════════════════════ + +BLOCKED_KEY_SUBSTRINGS: tuple[str, ...] = ( + "api_key", + "token", + "secret", + "password", + "client_secret", + "refresh_token", + "access_token", + "private_key", + "jwt", + "auth_token", + "credential", + "bearer", +) + +AUTH_BEARER_PATTERN = re.compile(r"Bearer\s+[A-Za-z0-9._~+/-]+", re.IGNORECASE) +ACCESS_TOKEN_QUERY_PATTERN = re.compile(r"(access_token=)[^&]+", re.IGNORECASE) +API_KEY_HEADER_PATTERN = re.compile(r"([A-Za-z0-9_-]*(?:api[_-]?key|token|secret|password)\s*[:=]\s*)[A-Za-z0-9._~+/-]+", re.IGNORECASE) +GENERIC_SECRET_PATTERNS = [ + re.compile(r"sk-[A-Za-z0-9_-]{10,}", re.IGNORECASE), + re.compile(r"gh[opsu]_[A-Za-z0-9_-]{10,}", re.IGNORECASE), + re.compile(r"xox[baprs]-[A-Za-z0-9_-]{10,}", re.IGNORECASE), + re.compile(r"eyJ[A-Za-z0-9_-]{10,}\.eyJ[A-Za-z0-9_-]{10,}\.[A-Za-z0-9._~+/-]{10,}", re.IGNORECASE), +] + + +def scrub_secrets(data: Any) -> Any: + """Recursively scrub credentials, API keys, and authorization tokens from strings, dictionaries, and collections.""" + if isinstance(data, dict): + result = {} + for k, v in data.items(): + k_lower = str(k).lower() + if any(sub in k_lower for sub in BLOCKED_KEY_SUBSTRINGS): + result[k] = "***" + elif isinstance(v, (dict, list, tuple, set)): + result[k] = scrub_secrets(v) + elif isinstance(v, str): + result[k] = scrub_string(v) + else: + result[k] = v + return result + elif isinstance(data, list): + return [scrub_secrets(item) for item in data] + elif isinstance(data, tuple): + return tuple(scrub_secrets(item) for item in data) + elif isinstance(data, set): + return {scrub_secrets(item) for item in data} + elif isinstance(data, str): + return scrub_string(data) + return data + + +def scrub_string(text: str) -> str: + """Mask credentials, bearer tokens, query parameters, and API keys within arbitrary text strings.""" + if not text or not isinstance(text, str): + return str(text or "") + s = AUTH_BEARER_PATTERN.sub("Bearer ***", text) + s = ACCESS_TOKEN_QUERY_PATTERN.sub(r"\1***", s) + s = API_KEY_HEADER_PATTERN.sub(r"\1***", s) + for pat in GENERIC_SECRET_PATTERNS: + s = pat.sub("***", s) + return s + + +# ═══════════════════════════════════════════════════════════════ +# P0-1: Workspace Boundary & Destructive Operations Guard +# ═══════════════════════════════════════════════════════════════ + +DESTRUCTIVE_COMMAND_NAMES: Set[str] = { + "rm", + "rmdir", + "unlink", + "del", + "erase", + "rd", + "remove-item", + "ri", + "trash-put", + "srm", + "shred", +} + + +class WorkspaceBoundaryGuard: + """Enforces explicit workspace boundaries, defends credential directories, and inspects destructive operations.""" + + def __init__(self, additional_allowed_roots: Optional[List[Path]] = None): + self._custom_roots: List[Path] = [r.resolve() for r in (additional_allowed_roots or [])] + + def get_allowed_roots(self) -> List[Path]: + """Return the canonical list of allowed roots (Project Root, HERMES_HOME, explicit allowed roots).""" + roots: List[Path] = [] + try: + repo_root = paths.get_repo_root().resolve() + roots.append(repo_root) + except Exception: + pass + + try: + hermes_home = paths.get_hermes_home().resolve() + roots.append(hermes_home) + except Exception: + pass + + roots.extend(self._custom_roots) + # Deduplicate while preserving order + deduped = [] + seen = set() + for r in roots: + resolved = r.resolve() + if resolved not in seen: + seen.add(resolved) + deduped.append(resolved) + return deduped + + def get_forbidden_paths(self) -> List[Path]: + """Return unconditionally protected paths (Credentials, SSH, Core Hub Configs).""" + forbidden = [] + try: + hermes_home = paths.get_hermes_home().resolve() + # Credential pools + forbidden.append((hermes_home / "agy_profiles").resolve()) + forbidden.append((hermes_home / "codex_profiles").resolve()) + forbidden.append((hermes_home / "opencode_profiles").resolve()) + forbidden.append((hermes_home / "claude_profiles").resolve()) + forbidden.append((hermes_home / "grok_profiles").resolve()) + forbidden.append((hermes_home / "local_profiles").resolve()) + # Sensitive config files + forbidden.append((hermes_home / "auth.json").resolve()) + forbidden.append((hermes_home / "hub_settings.json").resolve()) + forbidden.append((hermes_home / "router_profiles.yaml").resolve()) + except Exception: + pass + + # User SSH directory + try: + ssh_dir = (Path.home() / ".ssh").resolve() + forbidden.append(ssh_dir) + except Exception: + pass + + # User global credentials + try: + user_agy = (Path.home() / ".hermes" / "agy_profiles").resolve() + if user_agy not in forbidden: + forbidden.append(user_agy) + except Exception: + pass + + return forbidden + + 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() + for root in self.get_allowed_roots(): + try: + target.relative_to(root) + return True + except ValueError: + continue + return False + except Exception: + return False + + 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() + # 1. Exact match or child of forbidden directory + for fpath in self.get_forbidden_paths(): + if target == fpath: + return True, f"Путь {target} является защищённым системным ресурсом" + try: + target.relative_to(fpath) + return True, f"Путь {target} находится внутри защищённого каталога учётных данных {fpath}" + except ValueError: + continue + + # 2. Match critical filenames + target_name = target.name.lower() + if target_name in {"auth.json", "hub_settings.json", "id_rsa", "id_ed25519", "known_hosts"}: + return True, f"Файл {target_name} является защищённым системным файлом" + + # 3. Match .git core internals + for part in target.parts: + if part.lower() == ".git": + # Disallow deleting/modifying .git root, hooks, objects, or config + if target_name in {"config", "head", "index"} or "objects" in target.parts or "hooks" in target.parts: + return True, "Прямое разрушение служебных файлов git (.git/) запрещено" + + return False, None + except Exception as exc: + return True, f"Ошибка разрешения пути: {exc}" + + def validate_path(self, path: Path | str, operation: str = "read") -> Tuple[bool, str, Optional[str]]: + """Validate whether an operation on path is permitted. + + 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}", "Используйте стандартный относительный путь" + + # Check unconditional forbidden paths for mutating/deleting operations + if operation in {"delete", "write", "truncate", "move"}: + is_forbidden, forbidden_reason = self.is_forbidden_path(target) + if is_forbidden: + alt = ( + "Для удаления аккаунта используйте штатное действие 'delete_credentials' с подтверждением" + if "agy_profiles" in str(target) or "auth" in str(target) + else "Выполняйте операцию только в рабочей области проекта" + ) + return False, f"Запрещённая операция над защищённым ресурсом: {forbidden_reason}", alt + + # Check boundary containment + if not self.is_inside_allowed_root(target): + roots_display = ", ".join(str(r) for r in self.get_allowed_roots()) + return ( + False, + f"Путь '{target}' находится за пределами разрешённых рабочих областей: [{roots_display}]", + "Переместите целевой файл в каталог проекта или рабочую область агента", + ) + + return True, "OK", None + + def validate_command(self, cmd_line: str | List[str], cwd: Optional[Path | str] = None) -> Tuple[bool, str, Optional[str]]: + """Inspect and classify a shell command line for destructive or boundary-violating operations. + + Returns: (is_allowed: bool, reason: str, safe_alternative: Optional[str]) + """ + if not cmd_line: + return True, "OK", None + + # Parse command tokens + if isinstance(cmd_line, list): + tokens = list(cmd_line) + else: + try: + # Windows and POSIX-compatible shlex split + tokens = shlex.split(cmd_line, posix=(os.name != "nt")) + except Exception: + tokens = cmd_line.split() + + if not tokens: + return True, "OK", None + + base_cwd = Path(cwd or paths.get_repo_root()).resolve() + cmd_name = Path(tokens[0]).name.lower() + if cmd_name.endswith(".exe"): + cmd_name = cmd_name[:-4] + + # 1. Check destructive command names + if cmd_name in DESTRUCTIVE_COMMAND_NAMES: + # 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): + continue + targets.append(arg) + + if not targets: + # If no targets specified (e.g. interactive rm), check cwd + ok, reason, alt = self.validate_path(base_cwd, operation="delete") + if not ok: + return False, f"Команда {cmd_name} запущена в недопустимом каталоге: {reason}", alt + else: + for target_arg in targets: + target_path = (base_cwd / target_arg) if not Path(target_arg).is_absolute() else Path(target_arg) + ok, reason, alt = self.validate_path(target_path, operation="delete") + if not ok: + return False, f"Команда '{cmd_name}' пытается удалить недопустимый путь '{target_arg}': {reason}", alt + + # 2. Check git clean -fdx / git reset --hard targets + if cmd_name == "git": + subcmd = tokens[1].lower() if len(tokens) > 1 else "" + if subcmd == "clean" and any("-f" in a or "-x" in a or "-d" in a for a in tokens[2:]): + # Git clean inside workspace is allowed only if CWD is strictly within project root + ok, reason, alt = self.validate_path(base_cwd, operation="delete") + if not ok: + return False, f"git clean запущен вне проекта: {reason}", alt + + return True, "OK", None + + def dry_run_deletion(self, target_paths: List[Path | str]) -> Dict[str, Any]: + """Perform a safe dry-run assessment of a pending deletion, returning affected files and risk analysis.""" + files_to_delete: List[Dict[str, Any]] = [] + dirs_to_delete: List[Dict[str, Any]] = [] + total_bytes = 0 + risk_level = "low" + has_violations = False + reasons: List[str] = [] + + for p_raw in target_paths: + p = Path(p_raw).expanduser().resolve() + ok, reason, alt = self.validate_path(p, operation="delete") + if not ok: + has_violations = True + reasons.append(reason) + risk_level = "critical" + continue + + if p.is_file(): + try: + sz = p.stat().st_size + except OSError: + sz = 0 + total_bytes += sz + files_to_delete.append({"path": str(p), "size_bytes": sz, "type": "file"}) + elif p.is_dir(): + dir_bytes = 0 + file_count = 0 + for root, _, files in os.walk(p): + for f in files: + fp = Path(root) / f + try: + fsz = fp.stat().st_size + except OSError: + fsz = 0 + dir_bytes += fsz + file_count += 1 + total_bytes += dir_bytes + dirs_to_delete.append({"path": str(p), "file_count": file_count, "size_bytes": dir_bytes, "type": "directory"}) + if file_count > 10: + risk_level = "medium" if risk_level != "critical" else risk_level + + return { + "dry_run": True, + "allowed": not has_violations, + "risk_level": risk_level, + "total_files": len(files_to_delete), + "total_dirs": len(dirs_to_delete), + "total_bytes": total_bytes, + "files": files_to_delete[:100], + "directories": dirs_to_delete, + "violations": reasons, + } + + def safe_delete_file(self, path: Path | str, dry_run: bool = False) -> Dict[str, Any]: + """Safely delete a single file after boundary and credential checks, supporting dry-run.""" + target = Path(path).expanduser().resolve() + ok, reason, alt = self.validate_path(target, operation="delete") + if not ok: + raise BoundaryViolationError(reason, safe_alternative=alt) + + if dry_run: + return self.dry_run_deletion([target]) + + if not target.exists(): + return {"deleted": False, "reason": "File does not exist", "path": str(target)} + + if target.is_dir(): + raise BoundaryViolationError(f"Путь '{target}' является директорией, используйте safe_delete_dir") + + target.unlink() + return {"deleted": True, "path": str(target)} + + def safe_delete_dir(self, path: Path | str, dry_run: bool = False) -> Dict[str, Any]: + """Safely delete a directory tree after boundary and credential checks, supporting dry-run.""" + target = Path(path).expanduser().resolve() + ok, reason, alt = self.validate_path(target, operation="delete") + if not ok: + raise BoundaryViolationError(reason, safe_alternative=alt) + + if dry_run: + return self.dry_run_deletion([target]) + + if not target.exists(): + return {"deleted": False, "reason": "Directory does not exist", "path": str(target)} + + if not target.is_dir(): + raise BoundaryViolationError(f"Путь '{target}' не является директорией") + + shutil.rmtree(target) + return {"deleted": True, "path": str(target)} + + +# ═══════════════════════════════════════════════════════════════ +# P0-3: Network Boundary Guard & Whitelist +# ═══════════════════════════════════════════════════════════════ + +ALLOWED_OUTBOUND_HOSTS: Set[str] = { + # 1. AI Provider APIs + "api.anthropic.com", + "generativelanguage.googleapis.com", + "api.x.ai", + "api.openai.com", + "api.deepseek.com", + "openrouter.ai", + "api.together.xyz", + "api.groq.com", + "api.mistral.ai", + "dashscope.aliyuncs.com", + "open.bigmodel.cn", + "api.moonshot.cn", + # 2. Release & Update Hosts + "api.github.com", + "github.com", + "raw.githubusercontent.com", + "objects.githubusercontent.com", + "github-releases.githubusercontent.com", + # 3. Local LLMs & Diagnostics + "localhost", + "127.0.0.1", + "0.0.0.0", + "::1", +} + + +class NetworkBoundaryGuard: + """Enforces explicit network boundaries for all outgoing Hub HTTP/HTTPS requests.""" + + @classmethod + def is_host_allowed(cls, host_or_url: str) -> bool: + """Check whether the given hostname or URL target is allowed by the whitelist.""" + if not host_or_url: + return False + clean = host_or_url.strip().lower() + if "://" in clean: + parsed = urllib.parse.urlparse(clean) + host = (parsed.hostname or "").lower() + else: + host = clean.split(":")[0].lower() + + if not host: + return False + + # Exact match + if host in ALLOWED_OUTBOUND_HOSTS: + return True + + # Wildcard subdomain match (e.g. *.githubusercontent.com, *.aliyuncs.com, *.googleapis.com) + for allowed in ALLOWED_OUTBOUND_HOSTS: + if allowed.startswith("*."): + suffix = allowed[1:] + if host.endswith(suffix): + return True + elif host.endswith("." + allowed): + # Subdomains of explicitly trusted domains (e.g. download.github.com) + return True + + # Local subnet / private IP / loopback allow + if host.startswith("127.") or host.startswith("10.") or host.startswith("192.168."): + return True + + return False + + @classmethod + def validate_outbound_url(cls, url: str) -> None: + """Validate destination URL, raising NetworkBoundaryViolationError if host is not permitted.""" + if not cls.is_host_allowed(url): + parsed = urllib.parse.urlparse(url) + hostname = parsed.hostname or url + raise NetworkBoundaryViolationError( + f"Сетевое обращение к хосту '{hostname}' заблокировано сетевой границей хаба.", + safe_alternative="Используйте разрешённых провайдеров из белого списка или настройте локальный прокси", + ) + + +# ═══════════════════════════════════════════════════════════════ +# P0-2: Agent Workspace Guard & Role Credential Separation +# ═══════════════════════════════════════════════════════════════ + + +class AgentWorkspaceGuard: + """Provides workspace isolation and scoped credential injection per agent/role.""" + + @staticmethod + def get_agent_workspace_dir(agent_id: str) -> Path: + """Return dedicated workspace directory for an agent, ensuring it exists.""" + clean_id = re.sub(r"[^a-zA-Z0-9_-]+", "-", str(agent_id).strip()).strip("-").lower() + ws_dir = paths.get_hermes_home() / "workspaces" / f"agent-{clean_id}" + ws_dir.mkdir(parents=True, exist_ok=True) + return ws_dir.resolve() + + @staticmethod + def build_agent_subprocess_env( + agent_id: str, + role: str, + assigned_profile_id: Optional[str] = None, + ) -> Dict[str, str]: + """Construct an isolated environment containing ONLY the credentials needed for the agent's assigned role.""" + from antigravity_provider.agy_subprocess import build_safe_subprocess_env + from antigravity_provider.router.adapters.antigravity_adapter import get_profile_env_dir + + ws_dir = AgentWorkspaceGuard.get_agent_workspace_dir(agent_id) + overrides: Dict[str, str] = { + "HERMES_AGENT_ID": agent_id, + "HERMES_AGENT_ROLE": role, + "HERMES_AGENT_WORKSPACE": str(ws_dir), + } + + # Role-scoped profile isolation: inject HOME/USERPROFILE targeting the agent's profile directory + if assigned_profile_id: + profile_dir = get_profile_env_dir(assigned_profile_id) + overrides["HOME"] = str(profile_dir) + overrides["USERPROFILE"] = str(profile_dir) + overrides["HOMEPATH"] = str(profile_dir) + + return build_safe_subprocess_env(overrides=overrides) + + +# Singleton instance +_workspace_guard = WorkspaceBoundaryGuard() + + +def get_workspace_guard() -> WorkspaceBoundaryGuard: + return _workspace_guard diff --git a/src/antigravity_provider/router/unified_health.py b/src/antigravity_provider/router/unified_health.py index f6ee1dc..3b21b18 100644 --- a/src/antigravity_provider/router/unified_health.py +++ b/src/antigravity_provider/router/unified_health.py @@ -197,16 +197,22 @@ class SystemReadiness: # ═══════════════════════════════════════════════════════════════ -# Event Log Service +# ═══════════════════════════════════════════════════════════════ +# Event Log Service & Forensic Audit # ═══════════════════════════════════════════════════════════════ @dataclass class HubEvent: timestamp: str - category: str # account | quota | routing | auth | system + category: str # account | quota | routing | auth | system | security message: str details: Optional[str] = None level: str = "info" # info | warning | error | success + actor: str = "system" # user:web | reviewer:manual | agent: | system + action: Optional[str] = None + target_profile: Optional[str] = None + target_role: Optional[str] = None + outcome: str = "success" # success | denied | failed | dry_run class EventLogService: @@ -226,9 +232,36 @@ class EventLogService: cls._instance = cls() return cls._instance - def log(self, category: str, message: str, details: Optional[str] = None, level: str = "info"): + def log( + self, + category: str, + message: str, + details: Optional[str] = None, + level: str = "info", + actor: str = "system", + action: Optional[str] = None, + target_profile: Optional[str] = None, + target_role: Optional[str] = None, + outcome: str = "success", + ): + from antigravity_provider.router.security_guard import scrub_string + ts = time.strftime("%H:%M:%S") - event = HubEvent(timestamp=ts, category=category, message=message, details=details, level=level) + clean_msg = scrub_string(str(message)) + clean_details = scrub_string(str(details)) if details is not None else None + + event = HubEvent( + timestamp=ts, + category=category, + message=clean_msg, + details=clean_details, + level=level, + actor=str(actor or "system"), + action=str(action) if action else None, + target_profile=str(target_profile) if target_profile else None, + target_role=str(target_role) if target_role else None, + outcome=str(outcome or "success"), + ) with self._lock: self._events.append(event) # Cap at last 200 events @@ -251,8 +284,9 @@ class EventLogService: log_file = paths.get_log_file() clean_msg = sanitize_text(event.message) clean_details = sanitize_text(event.details) if event.details else None + actor_tag = f" [{event.actor}]" if event.actor != "system" else "" with open(log_file, "a", encoding="utf-8") as f: - f.write(f"[{event.timestamp}] [{event.category.upper()}] [{event.level.upper()}] {clean_msg}\n") + f.write(f"[{event.timestamp}] [{event.category.upper()}]{actor_tag} [{event.level.upper()}] {clean_msg}\n") if clean_details: f.write(f" Details: {clean_details}\n") except Exception: diff --git a/src/antigravity_provider/router/web/server.py b/src/antigravity_provider/router/web/server.py index 69569f6..48beabd 100644 --- a/src/antigravity_provider/router/web/server.py +++ b/src/antigravity_provider/router/web/server.py @@ -55,10 +55,11 @@ app = FastAPI(title="Hermes Hub Web API", version="1.0.0") # Собственному интерфейсу CORS не нужен: он отдаётся тем же сервером. Список # разрешённых источников оставлен настройкой — он понадобится, когда одна # панель будет смотреть на несколько хабов. +_raw_cors = str(_bootstrap_settings().get("web_api_allowed_origins", "")).split(",") _cors_origins = [ o.strip() - for o in str(_bootstrap_settings().get("web_api_allowed_origins", "")).split(",") - if o.strip() + for o in _raw_cors + if o.strip() and o.strip() != "*" ] if _cors_origins: app.add_middleware( @@ -66,7 +67,7 @@ if _cors_origins: allow_origins=_cors_origins, allow_credentials=True, allow_methods=["GET", "POST", "OPTIONS"], - allow_headers=["X-Hub-Token", "Content-Type"], + allow_headers=["X-Hub-Token", "Content-Type", "X-Hub-Actor"], ) @@ -194,6 +195,19 @@ def get_snapshot(authorized: bool = Depends(get_auth_token)): snap_dict = dataclasses.asdict(snapshot) snap_dict = sanitize_snapshot(snap_dict) + server_host = _web_settings().get("web_api_host", "127.0.0.1") + is_external = (server_host != "127.0.0.1" and server_host != "localhost") + snap_dict["network_security"] = { + "is_external_bind": is_external, + "is_tls": False, + "host": server_host, + "warning": ( + f"Внимание: Web API привязан к внешнему сетевому интерфейсу ({server_host}) поверх открытого HTTP. Токен авторизации и почты аккаунтов передаются по сети в открытом виде. Рекомендуется использовать HTTPS, VPN или SSH-туннель." + if is_external + else None + ), + } + return JSONResponse(content=jsonable_encoder(snap_dict)) @app.post("/api/action") @@ -210,7 +224,8 @@ async def handle_action(request: Request, authorized: bool = Depends(get_auth_to def _async_runner(func, name): threading.Thread(target=func, name=name, daemon=True).start() - result = ActionExecutor.execute(action, data.get("data", {}), async_runner=_async_runner) + actor = request.headers.get("X-Hub-Actor") or (f"web:{request.client.host}" if request.client else "user:web") + result = ActionExecutor.execute(action, data.get("data", {}), async_runner=_async_runner, actor=actor) if result.get("unknown"): raise HTTPException(status_code=404, detail="Неизвестное действие") @@ -237,8 +252,10 @@ def get_settings(authorized: bool = Depends(get_auth_token)): raw = _web_settings() has_token = bool(raw.get("web_api_token")) last_check = UpdateManager.get_last_check_result() + server_host = raw.get("web_api_host", "127.0.0.1") + is_external = (server_host != "127.0.0.1" and server_host != "localhost") settings_out: Dict[str, Any] = { - "web_api_host": raw.get("web_api_host", "127.0.0.1"), + "web_api_host": server_host, "web_api_port": raw.get("web_api_port", 5800), "web_api_token_configured": has_token, "theme": raw.get("theme", "system"), @@ -249,6 +266,16 @@ def get_settings(authorized: bool = Depends(get_auth_token)): "installed_commit": get_installed_commit(), "version": __version__, "last_update_check": last_check.to_dict() if last_check else None, + "network_security": { + "is_external_bind": is_external, + "is_tls": False, + "host": server_host, + "warning": ( + f"Внимание: Web API привязан к внешнему сетевому интерфейсу ({server_host}) поверх открытого HTTP. Токен авторизации и почты аккаунтов передаются по сети в открытом виде. Рекомендуется использовать HTTPS, VPN или SSH-туннель." + if is_external + else None + ), + }, } for k, v in raw.items(): if k not in settings_out and not any(secret in k.lower() for secret in ['token', 'secret', 'key', 'password', 'jwt']): diff --git a/tests/test_a37_isolation_guards.py b/tests/test_a37_isolation_guards.py new file mode 100644 index 0000000..216bad9 --- /dev/null +++ b/tests/test_a37_isolation_guards.py @@ -0,0 +1,368 @@ +"""Test suite for Task A37: Agent Isolation, Credential Protection, Workspace Boundaries, and Forensic Audit.""" +from __future__ import annotations + +import os +from pathlib import Path +from unittest.mock import patch +import pytest + +from fastapi.testclient import TestClient + +from antigravity_provider.router.security_guard import ( + WorkspaceBoundaryGuard, + NetworkBoundaryGuard, + AgentWorkspaceGuard, + BoundaryViolationError, + NetworkBoundaryViolationError, + scrub_secrets, + scrub_string, + get_workspace_guard, +) +from antigravity_provider.router.unified_health import EventLogService +from antigravity_provider.router.action_handler import ActionExecutor, do_delete_credentials +from antigravity_provider.router.web.server import app + + +# ═══════════════════════════════════════════════════════════════ +# P0-1: Workspace Boundary & Destructive Operations Guard Tests +# ═══════════════════════════════════════════════════════════════ + + +class TestWorkspaceBoundaryGuard: + """Tests for workspace boundaries, traversal prevention, and credential defense.""" + + def test_deletion_outside_allowed_workspace_is_blocked(self, tmp_path): + guard = WorkspaceBoundaryGuard(additional_allowed_roots=[tmp_path / "allowed_workspace"]) + (tmp_path / "allowed_workspace").mkdir(parents=True, exist_ok=True) + outside_file = tmp_path / "outside_workspace" / "sensitive_file.txt" + outside_file.parent.mkdir(parents=True, exist_ok=True) + outside_file.write_text("critical data", encoding="utf-8") + + ok, reason, safe_alt = guard.validate_path(outside_file, operation="delete") + assert ok is False + assert "за пределами разрешённых рабочих областей" in reason + assert safe_alt is not None + + with pytest.raises(BoundaryViolationError): + guard.safe_delete_file(outside_file) + + def test_unconditional_protection_of_credential_pools_and_ssh(self, tmp_path, monkeypatch): + hermes_home = tmp_path / ".hermes" + monkeypatch.setenv("HERMES_HOME", str(hermes_home)) + + guard = WorkspaceBoundaryGuard(additional_allowed_roots=[hermes_home]) + + # 1. agy_profiles + agy_profile = hermes_home / "agy_profiles" / "ag-w1" / "auth.json" + agy_profile.parent.mkdir(parents=True, exist_ok=True) + agy_profile.write_text("token", encoding="utf-8") + + ok, reason, alt = guard.validate_path(agy_profile, operation="delete") + assert ok is False + assert "защищённого каталога учётных данных" in reason or "защищённым" in reason + assert "delete_credentials" in alt + + # 2. hub_settings.json + settings_p = hermes_home / "hub_settings.json" + settings_p.write_text("{}", encoding="utf-8") + ok, reason, _ = guard.validate_path(settings_p, operation="delete") + assert ok is False + + # 3. .ssh + fake_ssh = Path.home() / ".ssh" / "id_rsa" + ok, reason, _ = guard.validate_path(fake_ssh, operation="delete") + assert ok is False + + def test_path_traversal_and_symlink_bypass_prevention(self, tmp_path): + ws_dir = tmp_path / "workspace" + ws_dir.mkdir(parents=True, exist_ok=True) + secret_outside = tmp_path / "secret.txt" + secret_outside.write_text("secret", encoding="utf-8") + + guard = WorkspaceBoundaryGuard(additional_allowed_roots=[ws_dir]) + + # Relative path traversal attack: workspace/../../secret.txt + traversal_path = ws_dir / "subdir" / ".." / ".." / "secret.txt" + ok, reason, _ = guard.validate_path(traversal_path, operation="delete") + assert ok is False + assert "за пределами" in reason + + # Symlink attack: symlink inside workspace pointing outside + link_inside = ws_dir / "link_to_outside" + try: + link_inside.symlink_to(secret_outside) + ok_symlink, reason_symlink, _ = guard.validate_path(link_inside, operation="delete") + assert ok_symlink is False + assert "за пределами" in reason_symlink + except OSError: + # On platforms without symlink privileges, skip symlink creation + pass + + def test_shell_command_analysis_and_classification(self, tmp_path): + guard = WorkspaceBoundaryGuard(additional_allowed_roots=[tmp_path / "project"]) + project_dir = tmp_path / "project" + project_dir.mkdir(parents=True, exist_ok=True) + + # 1. Dangerous rm -rf / or outside + ok, reason, _ = guard.validate_command("rm -rf /etc/passwd", cwd=project_dir) + assert ok is False + + # 2. Windows del C:\Windows + ok, reason, _ = guard.validate_command("del /f /q C:\\Windows\\System32", cwd=project_dir) + assert ok is False + + # 3. Safe rm within project + ok, reason, _ = guard.validate_command("rm temp_file.log", cwd=project_dir) + assert ok is True + + def test_dry_run_deletion_reports_metadata_without_deleting(self, tmp_path): + guard = WorkspaceBoundaryGuard(additional_allowed_roots=[tmp_path]) + test_dir = tmp_path / "to_delete" + test_dir.mkdir(parents=True, exist_ok=True) + file1 = test_dir / "file1.txt" + file2 = test_dir / "file2.txt" + file1.write_text("hello 1", encoding="utf-8") + file2.write_text("hello 222", encoding="utf-8") + + res = guard.dry_run_deletion([test_dir]) + assert res["dry_run"] is True + assert res["allowed"] is True + assert res["total_dirs"] >= 1 + assert res["total_files"] == 0 # directory item + assert res["total_bytes"] == len("hello 1".encode()) + len("hello 222".encode()) + assert file1.exists() + assert file2.exists() + + +# ═══════════════════════════════════════════════════════════════ +# P0-2: Agent Isolation & Credential Separation Tests +# ═══════════════════════════════════════════════════════════════ + + +class TestAgentIsolation: + """Tests for dedicated agent workspaces, role-scoped credentials, and actor attribution.""" + + def test_dedicated_agent_workspace_creation(self, tmp_path, monkeypatch): + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + + ws_dev1 = AgentWorkspaceGuard.get_agent_workspace_dir("developer-1") + ws_rev = AgentWorkspaceGuard.get_agent_workspace_dir("code-reviewer") + + assert ws_dev1.is_dir() + assert ws_rev.is_dir() + assert ws_dev1 != ws_rev + assert "agent-developer-1" in str(ws_dev1) + assert "agent-code-reviewer" in str(ws_rev) + + def test_role_scoped_credential_isolation(self, tmp_path, monkeypatch): + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + + env_dev1 = AgentWorkspaceGuard.build_agent_subprocess_env( + agent_id="developer-1", + role="coder", + assigned_profile_id="ag-w1", + ) + + assert env_dev1["HERMES_AGENT_ID"] == "developer-1" + assert env_dev1["HERMES_AGENT_ROLE"] == "coder" + assert "ag-w1" in env_dev1["HOME"] + assert "ag-w1" in env_dev1["USERPROFILE"] + + # Ensure no cross-agent foreign API keys leak + assert "ANTHROPIC_API_KEY" not in env_dev1 + assert "OPENAI_API_KEY" not in env_dev1 + + def test_actor_attribution_in_action_executor(self, tmp_path, monkeypatch): + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + + EventLogService._instance = None + event_svc = EventLogService.get() + + res = ActionExecutor.execute( + action="dry_run_delete", + data={"paths": [str(tmp_path)]}, + actor="agent:developer-1", + ) + + assert res["ok"] is True + events = event_svc.get_events(limit=5) + actor_events = [e for e in events if e.actor == "agent:developer-1"] + assert len(actor_events) > 0 + assert actor_events[0].action == "dry_run_delete" + assert actor_events[0].outcome == "dry_run" + + +# ═══════════════════════════════════════════════════════════════ +# P0-3: Network Boundary Guard & CORS Tests +# ═══════════════════════════════════════════════════════════════ + + +class TestNetworkBoundaryAndCORS: + """Tests for outbound destination whitelist, strict CORS, and network HTTP disclosures.""" + + def test_outbound_destination_whitelist(self): + # Allowed AI providers & releases + assert NetworkBoundaryGuard.is_host_allowed("https://api.anthropic.com/v1/messages") is True + assert NetworkBoundaryGuard.is_host_allowed("https://generativelanguage.googleapis.com") is True + assert NetworkBoundaryGuard.is_host_allowed("https://api.github.com/repos/releases") is True + assert NetworkBoundaryGuard.is_host_allowed("http://127.0.0.1:11434/api/tags") is True + assert NetworkBoundaryGuard.is_host_allowed("http://localhost:8000/v1/models") is True + + # Blocked external / rogue hosts + assert NetworkBoundaryGuard.is_host_allowed("https://evil-hacker.com/exfil") is False + assert NetworkBoundaryGuard.is_host_allowed("http://internal-artifactory.local:8081") is False + + with pytest.raises(NetworkBoundaryViolationError): + NetworkBoundaryGuard.validate_outbound_url("http://internal-artifactory.local/coordination") + + def test_cors_rejects_wildcard_with_credentials(self, client): + response = client.options( + "/api/snapshot", + headers={"Origin": "https://attacker.example.com", "Access-Control-Request-Method": "GET"}, + ) + # Should NOT return allow-origin: * or echo back attacker origin + allow_origin = response.headers.get("access-control-allow-origin") + assert allow_origin != "*" + assert allow_origin != "https://attacker.example.com" + + def test_network_http_warning_on_external_bind(self, client, monkeypatch): + from antigravity_provider.router import settings_service + + fake_settings = { + "web_api_host": "0.0.0.0", + "web_api_token": "secret_token_123", + } + monkeypatch.setattr(settings_service, "get_hub_settings", lambda: fake_settings) + + res = client.get("/api/settings", headers={"X-Hub-Token": "secret_token_123"}) + assert res.status_code == 200 + data = res.json() + assert "network_security" in data + assert data["network_security"]["is_external_bind"] is True + assert "открытого HTTP" in data["network_security"]["warning"] + + +# ═══════════════════════════════════════════════════════════════ +# P0-4: Forensic Audit Logging & Secret Scrubbing Tests +# ═══════════════════════════════════════════════════════════════ + + +class TestForensicAuditAndSecretScrubbing: + """Tests for secret scrubbing and forensic traceability.""" + + def test_secret_scrubbing_removes_tokens_and_bearer_headers(self): + dirty = { + "user_prompt": "Please summarize this commit", + "auth_header": "Bearer sk-proj-1234567890abcdef123456", + "api_key": "secret_api_key_value", + "nested": { + "token": "ghp_12345678901234567890", + "safe_field": "visible value", + }, + } + + clean = scrub_secrets(dirty) + assert clean["api_key"] == "***" + assert clean["nested"]["token"] == "***" + assert clean["nested"]["safe_field"] == "visible value" + assert "sk-proj" not in clean["auth_header"] + + def test_string_scrubber_masks_inline_secrets(self): + msg = "Request failed with error: Bearer ghp_abcdef1234567890 on host https://api.openai.com?access_token=sk-9876543210" + scrubbed = scrub_string(msg) + assert "ghp_abcdef" not in scrubbed + assert "sk-987654" not in scrubbed + assert "Bearer ***" in scrubbed + assert "access_token=***" in scrubbed + + def test_audit_log_captures_actor_and_outcome(self, tmp_path, monkeypatch): + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + EventLogService._instance = None + svc = EventLogService.get() + + svc.log( + category="account", + message="Удаление профиля grok-worker-1", + actor="reviewer:manual", + action="delete_credentials", + target_profile="grok-worker-1", + target_role="code-reviewer", + outcome="success", + level="warning", + ) + + events = svc.get_events(limit=5) + assert len(events) >= 1 + ev = events[0] + assert ev.actor == "reviewer:manual" + assert ev.action == "delete_credentials" + assert ev.target_profile == "grok-worker-1" + assert ev.target_role == "code-reviewer" + assert ev.outcome == "success" + + +# ═══════════════════════════════════════════════════════════════ +# P0-5: Live Deletion Confirmation & Safe Operations Tests +# ═══════════════════════════════════════════════════════════════ + + +class TestCredentialsDeletionGuard: + """Tests for explicit credential deletion confirmation and dry-run.""" + + def test_delete_credentials_dry_run(self, tmp_path, monkeypatch): + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + from antigravity_provider.router.profile_manager import get_profile_auth_path + + auth_p = get_profile_auth_path("grok", "grok-worker-1") + auth_p.parent.mkdir(parents=True, exist_ok=True) + auth_p.write_text("{\"key\": \"val\"}", encoding="utf-8") + + res = ActionExecutor.execute( + action="delete_credentials", + data={"provider": "grok", "profile_id": "grok-worker-1", "dry_run": True}, + actor="reviewer:manual", + ) + + assert res["ok"] is True + assert res.get("dry_run") is True + assert auth_p.is_file() # File MUST NOT be deleted during dry-run + + def test_delete_credentials_unconfirmed_request(self, tmp_path, monkeypatch): + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + from antigravity_provider.router.profile_manager import get_profile_auth_path + + auth_p = get_profile_auth_path("grok", "grok-worker-1") + auth_p.parent.mkdir(parents=True, exist_ok=True) + auth_p.write_text("{\"key\": \"val\"}", encoding="utf-8") + + res = ActionExecutor.execute( + action="delete_credentials", + data={"provider": "grok", "profile_id": "grok-worker-1", "confirmed": False}, + actor="agent:developer-1", + ) + + assert res["ok"] is False + assert res.get("confirmation_required") is True + assert auth_p.is_file() # File MUST NOT be deleted without confirmation + + def test_delete_credentials_confirmed_success(self, tmp_path, monkeypatch): + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + from antigravity_provider.router.profile_manager import get_profile_auth_path + + auth_p = get_profile_auth_path("grok", "grok-worker-1") + auth_p.parent.mkdir(parents=True, exist_ok=True) + auth_p.write_text("{\"key\": \"val\"}", encoding="utf-8") + + res = ActionExecutor.execute( + action="delete_credentials", + data={"provider": "grok", "profile_id": "grok-worker-1", "confirmed": True}, + actor="user:web", + ) + + assert res["ok"] is True + assert not auth_p.is_file() + + +@pytest.fixture +def client(): + return TestClient(app)