refactor(profiles): extract the safe tar.gz primitives into archive_safe
Profile export/import owns the only hardened tar handling in the tree: GNU-format writing (PAX fractional mtimes make macOS Archive Utility throw "Error 94"), plus an extractor that rejects absolute paths, `..` components, and non-regular members. Kanban board transfer needs exactly that, and a second copy is how the weaker of two extractors eventually ships. Move the four helpers to hermes_cli/archive_safe and point profiles at them; no behavior change beyond dropping a provably-unreachable fallback in the root-listing helper, whose condition is a strict subset of the comprehension above it.
This commit is contained in:
Brooklyn Nicholson
committed by
brooklyn!
brooklyn!
parent
9a1eef7a29
commit
110ecd238e
130
hermes_cli/archive_safe.py
Normal file
130
hermes_cli/archive_safe.py
Normal file
@@ -0,0 +1,130 @@
|
||||
"""Safe ``tar.gz`` primitives shared by the profile and kanban transfer paths.
|
||||
|
||||
Both ``hermes profile export|import`` and ``hermes kanban export|import``
|
||||
ship a directory to another machine and unpack whatever comes back. The
|
||||
unpack side is the dangerous half: a hand-crafted archive can carry
|
||||
``../`` members, absolute paths, symlinks, or device nodes, any of which
|
||||
turn an import into an arbitrary-write primitive. These helpers are the
|
||||
one place that logic lives so a second transfer surface can't ship a
|
||||
second, subtly weaker extractor.
|
||||
|
||||
The writer is deliberately not :func:`shutil.make_archive`: that emits
|
||||
PAX (Python's tarfile default since 3.8), whose fractional-mtime records
|
||||
macOS Archive Utility rejects — double-clicking an exported profile threw
|
||||
"Error 94 - Bad message." GNU format keeps long paths working (longlink
|
||||
extensions) and stays integer-mtime, so Finder, bsdtar, and gnutar all
|
||||
extract it.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import shutil
|
||||
import tarfile
|
||||
from pathlib import Path, PurePosixPath, PureWindowsPath
|
||||
|
||||
|
||||
def normalize_archive_parts(member_name: str) -> list[str]:
|
||||
"""Return safe path parts for an archive member, or raise.
|
||||
|
||||
Rejects absolute paths (POSIX and Windows, including drive letters),
|
||||
empty names, and any ``..`` component. Backslashes are folded to
|
||||
``/`` first so a Windows-authored archive can't smuggle a separator
|
||||
past the POSIX parse.
|
||||
"""
|
||||
normalized_name = member_name.replace("\\", "/")
|
||||
posix_path = PurePosixPath(normalized_name)
|
||||
windows_path = PureWindowsPath(member_name)
|
||||
|
||||
if (
|
||||
not normalized_name
|
||||
or posix_path.is_absolute()
|
||||
or windows_path.is_absolute()
|
||||
or windows_path.drive
|
||||
):
|
||||
raise ValueError(f"Unsafe archive member path: {member_name}")
|
||||
|
||||
parts = [part for part in posix_path.parts if part not in {"", "."}]
|
||||
if not parts or any(part == ".." for part in parts):
|
||||
raise ValueError(f"Unsafe archive member path: {member_name}")
|
||||
return parts
|
||||
|
||||
|
||||
def make_targz(base: str, root_dir: str, base_dir: str) -> str:
|
||||
"""Create ``<base>.tar.gz`` of ``root_dir/base_dir`` in GNU tar format."""
|
||||
archive_path = f"{base}.tar.gz"
|
||||
with tarfile.open(archive_path, "w:gz", format=tarfile.GNU_FORMAT) as tf:
|
||||
tf.add(str(Path(root_dir) / base_dir), arcname=base_dir)
|
||||
return archive_path
|
||||
|
||||
|
||||
def safe_extract_targz(archive: Path, destination: Path) -> None:
|
||||
"""Extract ``archive`` into ``destination`` without path escapes or links.
|
||||
|
||||
Only directories and regular files are extracted; symlinks, hardlinks,
|
||||
and device nodes raise rather than being silently skipped, so a
|
||||
tampered archive fails the import instead of landing a partial tree.
|
||||
"""
|
||||
with tarfile.open(archive, "r:gz") as tf:
|
||||
for member in tf.getmembers():
|
||||
parts = normalize_archive_parts(member.name)
|
||||
target = destination.joinpath(*parts)
|
||||
|
||||
if member.isdir():
|
||||
target.mkdir(parents=True, exist_ok=True)
|
||||
continue
|
||||
|
||||
if not member.isfile():
|
||||
raise ValueError(
|
||||
f"Unsupported archive member type: {member.name}"
|
||||
)
|
||||
|
||||
target.parent.mkdir(parents=True, exist_ok=True)
|
||||
extracted = tf.extractfile(member)
|
||||
if extracted is None:
|
||||
raise ValueError(f"Cannot read archive member: {member.name}")
|
||||
|
||||
with extracted, open(target, "wb") as dst:
|
||||
shutil.copyfileobj(extracted, dst)
|
||||
|
||||
try:
|
||||
os.chmod(target, member.mode & 0o777)
|
||||
except OSError:
|
||||
pass
|
||||
|
||||
|
||||
def archive_root_dirs(archive: Path) -> set[str]:
|
||||
"""Return the archive's top-level directory names.
|
||||
|
||||
Transfer archives carry exactly one root directory, which names the
|
||||
thing being imported. Inspecting the archive before extraction lets
|
||||
the caller resolve the target name (and refuse a malformed archive)
|
||||
without first mutating a live tree.
|
||||
"""
|
||||
with tarfile.open(archive, "r:gz") as tf:
|
||||
return {
|
||||
parts[0]
|
||||
for member in tf.getmembers()
|
||||
for parts in [normalize_archive_parts(member.name)]
|
||||
if len(parts) > 1 or member.isdir()
|
||||
}
|
||||
|
||||
|
||||
def copy_regular_files(src: Path, dst: Path) -> int:
|
||||
"""Copy the regular files under ``src`` into ``dst``, skipping symlinks.
|
||||
|
||||
Used on the *export* side so a symlink planted in an attachments or
|
||||
logs tree can't pull an arbitrary file into the archive. Returns the
|
||||
number of files copied; a missing ``src`` copies nothing.
|
||||
"""
|
||||
if not src.is_dir():
|
||||
return 0
|
||||
copied = 0
|
||||
for entry in sorted(src.rglob("*")):
|
||||
if entry.is_symlink() or not entry.is_file():
|
||||
continue
|
||||
target = dst / entry.relative_to(src)
|
||||
target.parent.mkdir(parents=True, exist_ok=True)
|
||||
shutil.copyfile(entry, target)
|
||||
copied += 1
|
||||
return copied
|
||||
@@ -30,10 +30,16 @@ import subprocess
|
||||
import sys
|
||||
import time
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path, PurePosixPath, PureWindowsPath
|
||||
from pathlib import Path
|
||||
from typing import Dict, List, Optional, Tuple
|
||||
|
||||
from agent.skill_utils import is_excluded_skill_path
|
||||
from hermes_cli.archive_safe import (
|
||||
archive_root_dirs,
|
||||
make_targz,
|
||||
normalize_archive_parts,
|
||||
safe_extract_targz,
|
||||
)
|
||||
from hermes_constants import (
|
||||
clear_named_profile_deleted,
|
||||
mark_named_profile_deleted,
|
||||
@@ -2134,23 +2140,6 @@ def _default_export_ignore(root_dir: Path):
|
||||
return _ignore
|
||||
|
||||
|
||||
def _make_profile_archive(base: str, root_dir: str, base_dir: str) -> str:
|
||||
"""Create ``<base>.tar.gz`` of ``root_dir/base_dir`` — GNU tar format.
|
||||
|
||||
Not :func:`shutil.make_archive`: that writes PAX (Python's tarfile default
|
||||
since 3.8), whose fractional-mtime records macOS Archive Utility rejects —
|
||||
double-clicking an exported profile threw "Error 94 - Bad message." GNU
|
||||
format keeps long paths working (longlink extensions) and stays integer-
|
||||
mtime, so Finder, bsdtar, and gnutar all extract it.
|
||||
"""
|
||||
import tarfile
|
||||
|
||||
archive_path = f"{base}.tar.gz"
|
||||
with tarfile.open(archive_path, "w:gz", format=tarfile.GNU_FORMAT) as tf:
|
||||
tf.add(str(Path(root_dir) / base_dir), arcname=base_dir)
|
||||
return archive_path
|
||||
|
||||
|
||||
# Text / config suffixes walked during export secret scrubbing. Binary DBs,
|
||||
# images, and other non-text artifacts are left alone (they may still leave
|
||||
# via named-profile export — scrubbing those is a separate concern).
|
||||
@@ -2249,7 +2238,7 @@ def export_profile(name: str, output_path: str, extra_files: Optional[Dict[str,
|
||||
|
||||
def _stage_extras(staged: Path) -> None:
|
||||
for rel, content in (extra_files or {}).items():
|
||||
parts = _normalize_profile_archive_parts(rel)
|
||||
parts = normalize_archive_parts(rel)
|
||||
target = staged.joinpath(*parts)
|
||||
target.parent.mkdir(parents=True, exist_ok=True)
|
||||
target.write_text(content, encoding="utf-8")
|
||||
@@ -2268,7 +2257,7 @@ def export_profile(name: str, output_path: str, extra_files: Optional[Dict[str,
|
||||
)
|
||||
_stage_extras(staged)
|
||||
_scrub_export_secrets(staged)
|
||||
result = _make_profile_archive(base, tmpdir, "default")
|
||||
result = make_targz(base, tmpdir, "default")
|
||||
return Path(result)
|
||||
|
||||
# Named profiles — stage a filtered copy to exclude credentials
|
||||
@@ -2283,87 +2272,10 @@ def export_profile(name: str, output_path: str, extra_files: Optional[Dict[str,
|
||||
)
|
||||
_stage_extras(staged)
|
||||
_scrub_export_secrets(staged)
|
||||
result = _make_profile_archive(base, tmpdir, canon)
|
||||
result = make_targz(base, tmpdir, canon)
|
||||
return Path(result)
|
||||
|
||||
|
||||
def _normalize_profile_archive_parts(member_name: str) -> List[str]:
|
||||
"""Return safe path parts for a profile archive member."""
|
||||
normalized_name = member_name.replace("\\", "/")
|
||||
posix_path = PurePosixPath(normalized_name)
|
||||
windows_path = PureWindowsPath(member_name)
|
||||
|
||||
if (
|
||||
not normalized_name
|
||||
or posix_path.is_absolute()
|
||||
or windows_path.is_absolute()
|
||||
or windows_path.drive
|
||||
):
|
||||
raise ValueError(f"Unsafe archive member path: {member_name}")
|
||||
|
||||
parts = [part for part in posix_path.parts if part not in {"", "."}]
|
||||
if not parts or any(part == ".." for part in parts):
|
||||
raise ValueError(f"Unsafe archive member path: {member_name}")
|
||||
return parts
|
||||
|
||||
|
||||
def _safe_extract_profile_archive(archive: Path, destination: Path) -> None:
|
||||
"""Extract a profile archive without allowing path escapes or links."""
|
||||
import tarfile
|
||||
|
||||
with tarfile.open(archive, "r:gz") as tf:
|
||||
for member in tf.getmembers():
|
||||
parts = _normalize_profile_archive_parts(member.name)
|
||||
target = destination.joinpath(*parts)
|
||||
|
||||
if member.isdir():
|
||||
target.mkdir(parents=True, exist_ok=True)
|
||||
continue
|
||||
|
||||
if not member.isfile():
|
||||
raise ValueError(
|
||||
f"Unsupported archive member type: {member.name}"
|
||||
)
|
||||
|
||||
target.parent.mkdir(parents=True, exist_ok=True)
|
||||
extracted = tf.extractfile(member)
|
||||
if extracted is None:
|
||||
raise ValueError(f"Cannot read archive member: {member.name}")
|
||||
|
||||
with extracted, open(target, "wb") as dst:
|
||||
shutil.copyfileobj(extracted, dst)
|
||||
|
||||
try:
|
||||
os.chmod(target, member.mode & 0o777)
|
||||
except OSError:
|
||||
pass
|
||||
|
||||
|
||||
def _inspect_profile_archive_roots(archive: Path) -> set[str]:
|
||||
"""Return the archive's top-level directory names.
|
||||
|
||||
Profile imports expect exactly one root directory. Inspecting the archive
|
||||
before extraction lets us stage the import safely instead of mutating a
|
||||
live profile tree first and reconciling names later.
|
||||
"""
|
||||
import tarfile
|
||||
|
||||
with tarfile.open(archive, "r:gz") as tf:
|
||||
top_dirs = {
|
||||
parts[0]
|
||||
for member in tf.getmembers()
|
||||
for parts in [_normalize_profile_archive_parts(member.name)]
|
||||
if len(parts) > 1 or member.isdir()
|
||||
}
|
||||
if not top_dirs:
|
||||
top_dirs = {
|
||||
_normalize_profile_archive_parts(member.name)[0]
|
||||
for member in tf.getmembers()
|
||||
if member.isdir()
|
||||
}
|
||||
return top_dirs
|
||||
|
||||
|
||||
def import_profile(archive_path: str, name: Optional[str] = None) -> Path:
|
||||
"""Import a profile from a tar.gz archive.
|
||||
|
||||
@@ -2376,7 +2288,7 @@ def import_profile(archive_path: str, name: Optional[str] = None) -> Path:
|
||||
if not archive.exists():
|
||||
raise FileNotFoundError(f"Archive not found: {archive}")
|
||||
|
||||
top_dirs = _inspect_profile_archive_roots(archive)
|
||||
top_dirs = archive_root_dirs(archive)
|
||||
archive_root = top_dirs.pop() if len(top_dirs) == 1 else None
|
||||
inferred_name = name or archive_root
|
||||
if not inferred_name:
|
||||
@@ -2409,7 +2321,7 @@ def import_profile(archive_path: str, name: Optional[str] = None) -> Path:
|
||||
|
||||
with tempfile.TemporaryDirectory(prefix="hermes_profile_import_") as tmpdir:
|
||||
staging_root = Path(tmpdir)
|
||||
_safe_extract_profile_archive(archive, staging_root)
|
||||
safe_extract_targz(archive, staging_root)
|
||||
|
||||
extracted = staging_root / archive_root
|
||||
if not extracted.is_dir():
|
||||
|
||||
Reference in New Issue
Block a user