128 lines
4.1 KiB
Python
128 lines
4.1 KiB
Python
"""Principal profile mapping helpers.
|
|
|
|
Read docs/profile-isolation/README.md and docs/profile-isolation/checklist.md
|
|
before changing how principals map to profile directories.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import logging
|
|
from pathlib import Path
|
|
from tempfile import NamedTemporaryFile
|
|
from typing import Any, Optional
|
|
|
|
from hermes_cli.profiles import create_profile, get_profile_dir, profile_exists
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def default_principal_profile_map_path() -> Path:
|
|
"""Return the host-global principal→profile mapping file path.
|
|
|
|
Anchored to the built-in default Hermes home so all gateway profiles would
|
|
share the same mapping if/when multiplexing expands beyond the default
|
|
profile.
|
|
"""
|
|
from hermes_cli.profiles import get_profile_dir as _get_profile_dir
|
|
|
|
return _get_profile_dir("default") / "principal_profile_map.json"
|
|
|
|
|
|
def principal_profile_name(principal_id: str) -> str:
|
|
principal = str(principal_id or "").strip().lower()
|
|
if not principal:
|
|
raise ValueError("principal_id is required")
|
|
slug = "".join(ch for ch in principal if ch.isalnum())[:8]
|
|
if not slug:
|
|
raise ValueError("principal_id must contain at least one alphanumeric character")
|
|
return f"principal_{slug}"
|
|
|
|
|
|
def _load_mapping(map_path: Path) -> dict[str, str]:
|
|
if not map_path.exists():
|
|
return {}
|
|
try:
|
|
raw = json.loads(map_path.read_text(encoding="utf-8"))
|
|
except Exception:
|
|
logger.warning("Failed to read principal profile map from %s", map_path, exc_info=True)
|
|
return {}
|
|
if not isinstance(raw, dict):
|
|
return {}
|
|
return {str(k): str(v) for k, v in raw.items() if k and v}
|
|
|
|
|
|
def _write_mapping(map_path: Path, mapping: dict[str, str]) -> None:
|
|
map_path.parent.mkdir(parents=True, exist_ok=True)
|
|
with NamedTemporaryFile("w", encoding="utf-8", dir=str(map_path.parent), delete=False) as tmp:
|
|
json.dump(mapping, tmp, indent=2, sort_keys=True, ensure_ascii=False)
|
|
tmp.write("\n")
|
|
tmp_path = Path(tmp.name)
|
|
tmp_path.replace(map_path)
|
|
|
|
|
|
def ensure_principal_profile(
|
|
principal_id: str,
|
|
*,
|
|
map_path: Optional[Path] = None,
|
|
create_profile_on_missing: bool = True,
|
|
) -> str:
|
|
principal = str(principal_id or "").strip()
|
|
if not principal:
|
|
raise ValueError("principal_id is required")
|
|
target_map_path = Path(map_path) if map_path else default_principal_profile_map_path()
|
|
mapping = _load_mapping(target_map_path)
|
|
profile_name = mapping.get(principal) or principal_profile_name(principal)
|
|
if mapping.get(principal) != profile_name:
|
|
mapping[principal] = profile_name
|
|
_write_mapping(target_map_path, mapping)
|
|
if create_profile_on_missing and not profile_exists(profile_name):
|
|
try:
|
|
create_profile(profile_name, no_alias=True)
|
|
except FileExistsError:
|
|
pass
|
|
except Exception:
|
|
logger.warning(
|
|
"Failed to auto-create principal profile %s for %s",
|
|
profile_name,
|
|
principal,
|
|
exc_info=True,
|
|
)
|
|
raise
|
|
return profile_name
|
|
|
|
|
|
def resolve_principal_profile(
|
|
source: Any,
|
|
*,
|
|
verification_store: Any,
|
|
map_path: Optional[Path] = None,
|
|
create_profile: bool = True,
|
|
) -> str | None:
|
|
if source is None or verification_store is None:
|
|
return None
|
|
if getattr(source, "profile", None):
|
|
return str(source.profile)
|
|
get_context = getattr(verification_store, "get_principal_context", None)
|
|
if get_context is None:
|
|
return None
|
|
try:
|
|
context = get_context(source)
|
|
except Exception:
|
|
logger.debug("principal context lookup failed", exc_info=True)
|
|
return None
|
|
if not isinstance(context, dict):
|
|
return None
|
|
principal_id = context.get("principal_id")
|
|
if not principal_id:
|
|
return None
|
|
return ensure_principal_profile(
|
|
str(principal_id),
|
|
map_path=Path(map_path) if map_path else None,
|
|
create_profile_on_missing=create_profile,
|
|
)
|
|
|
|
|
|
def principal_profile_home(profile_name: str) -> Path:
|
|
return get_profile_dir(profile_name)
|