memory.safety

memory.safety

Shared safety primitives for the angelo backend.

This module is intentionally stdlib-only so that every subsystem (memory, zettelkasten, coordinator, angelo_cli, dashboards, artifacts) can import it without creating package coupling or pulling in heavy optional dependencies.

It provides the building blocks that the hardening work depends on:

• :func:validate_id -- containment check for IDs/names used to build paths.
• :func:safe_join -- join + resolve + assert the result stays under a root.
• :func:is_sensitive_path -- denylist for files that must never be committed/pushed.
• :func:atomic_write -- crash-safe write (temp file + fsync + os.replace).
• :func:file_lock -- best-effort cross-process advisory lock.

Design notes

• validate_id is a containment check, NOT a format validator. It rejects path-traversal characters only, so existing mixed-format IDs (anno-v2security, chec-memory-recovery-20260610, namespaced ext:note-1) all remain valid.
• The real containment guarantee comes from :func:safe_join, which resolves the final path and asserts it is relative to the root (using is_relative_to rather than the bypassable str().startswith() pattern).

Nothing in this module is wired into existing code paths yet; it is pure additive code intended to be adopted call site by call site.

validate_id

validate_id(value: str, *, kind: str = 'id', for_filename: bool = False) -> str

Validate that value is safe to interpolate into a filesystem path.

By default this is a containment check, not a format check. It rejects empty values, path separators, .. sequences, absolute paths, control characters, and over-long strings -- but otherwise permits any characters (letters, digits, -, _, ., : for namespaced IDs, etc.) so existing IDs keep working.

When for_filename=True the check is stricter: it additionally rejects Windows-illegal filename characters (including :, which would otherwise create an NTFS alternate data stream), trailing dots/spaces (silently stripped by Windows), and reserved device names. Use this mode at sites that turn the value directly into a filename.

Returns the value unchanged when valid; raises :class:ValueError otherwise.

Source code in memory/safety.py

def validate_id(value: str, *, kind: str = "id", for_filename: bool = False) -> str:
    """Validate that ``value`` is safe to interpolate into a filesystem path.

    By default this is a *containment* check, not a format check. It rejects
    empty values, path separators, ``..`` sequences, absolute paths, control
    characters, and over-long strings -- but otherwise permits any characters
    (letters, digits, ``-``, ``_``, ``.``, ``:`` for namespaced IDs, etc.) so
    existing IDs keep working.

    When ``for_filename=True`` the check is stricter: it additionally rejects
    Windows-illegal filename characters (including ``:``, which would otherwise
    create an NTFS alternate data stream), trailing dots/spaces (silently
    stripped by Windows), and reserved device names. Use this mode at sites that
    turn the value directly into a filename.

    Returns the value unchanged when valid; raises :class:`ValueError` otherwise.
    """
    if not isinstance(value, str):
        raise ValueError(f"{kind} must be a string, got {type(value).__name__}")
    if not value:
        raise ValueError(f"{kind} must not be empty")
    if len(value) > _MAX_ID_LEN:
        raise ValueError(f"{kind} is too long ({len(value)} > {_MAX_ID_LEN} chars)")
    for ch in _ID_FORBIDDEN_CHARS:
        if ch in value:
            raise ValueError(f"{kind} must not contain {ch!r}")
    if ".." in value:
        raise ValueError(f"{kind} must not contain '..'")
    if value in (".", ".."):
        raise ValueError(f"{kind} must not be a relative path component")
    if any(ord(c) < 32 for c in value):
        raise ValueError(f"{kind} must not contain control characters")
    if os.path.isabs(value):
        raise ValueError(f"{kind} must not be an absolute path")

    if for_filename:
        for ch in _WINDOWS_ILLEGAL_FILENAME_CHARS:
            if ch in value:
                raise ValueError(
                    f"{kind} must not contain {ch!r} when used as a filename"
                )
        if value[-1] in (" ", "."):
            raise ValueError(f"{kind} must not end with a space or '.'")
        stem = value.split(".", 1)[0].lower()
        if stem in _WINDOWS_RESERVED_NAMES:
            raise ValueError(f"{kind} must not be a reserved device name")

    return value

safe_join

safe_join(root: Union[str, PathLike], *parts: str) -> Path

Join parts onto root and guarantee the result stays inside root.

Resolves both the root and the candidate path, then asserts the candidate is relative to the resolved root. This catches .. traversal, absolute-path escapes, and (because it resolves symlinks) symlink escapes -- unlike the str(resolved).startswith(str(root)) pattern used elsewhere, which is vulnerable to sibling-prefix and casing tricks.

Returns the resolved :class:~pathlib.Path; raises :class:ValueError if the candidate would escape root.

Source code in memory/safety.py

def safe_join(root: Union[str, os.PathLike], *parts: str) -> Path:
    """Join ``parts`` onto ``root`` and guarantee the result stays inside ``root``.

    Resolves both the root and the candidate path, then asserts the candidate is
    relative to the resolved root. This catches ``..`` traversal, absolute-path
    escapes, and (because it resolves symlinks) symlink escapes -- unlike the
    ``str(resolved).startswith(str(root))`` pattern used elsewhere, which is
    vulnerable to sibling-prefix and casing tricks.

    Returns the resolved :class:`~pathlib.Path`; raises :class:`ValueError` if the
    candidate would escape ``root``.
    """
    root_resolved = Path(root).resolve()
    candidate = root_resolved.joinpath(*parts).resolve()
    if candidate != root_resolved and not _is_relative_to(candidate, root_resolved):
        raise ValueError(
            f"path {candidate!s} escapes root {root_resolved!s}"
        )
    return candidate

is_sensitive_path

is_sensitive_path(path: Union[str, PathLike]) -> bool

Return True if path looks like a secret/credential file.

Used as a denylist (never an allowlist) so that ordinary code/data files remain eligible for git pinning and DVC tracking -- only credential-bearing files are excluded.

Source code in memory/safety.py

def is_sensitive_path(path: Union[str, os.PathLike]) -> bool:
    """Return ``True`` if ``path`` looks like a secret/credential file.

    Used as a denylist (never an allowlist) so that ordinary code/data files
    remain eligible for git pinning and DVC tracking -- only credential-bearing
    files are excluded.
    """
    p = Path(path)
    parts_lower = [part.lower() for part in p.parts]
    if any(part in _SENSITIVE_DIR_NAMES for part in parts_lower):
        return True
    name = p.name.lower()
    for pattern in _SENSITIVE_NAME_GLOBS:
        if fnmatch.fnmatch(name, pattern):
            return True
    return False

atomic_write

atomic_write(path: Union[str, PathLike], data: Union[str, bytes], *, encoding: str = 'utf-8') -> Path

Write data to path atomically.

Writes to a temporary file in the same directory, flushes and fsyncs it, then uses :func:os.replace for an atomic rename. A crash mid-write leaves either the old file or the new file -- never a truncated one. Parent directories are created as needed.

Returns the destination :class:~pathlib.Path.

Source code in memory/safety.py

def atomic_write(
    path: Union[str, os.PathLike],
    data: Union[str, bytes],
    *,
    encoding: str = "utf-8",
) -> Path:
    """Write ``data`` to ``path`` atomically.

    Writes to a temporary file in the same directory, flushes and fsyncs it, then
    uses :func:`os.replace` for an atomic rename. A crash mid-write leaves either
    the old file or the new file -- never a truncated one. Parent directories are
    created as needed.

    Returns the destination :class:`~pathlib.Path`.
    """
    dest = Path(path)
    dest.parent.mkdir(parents=True, exist_ok=True)

    if isinstance(data, str):
        raw = data.encode(encoding)
    else:
        raw = data

    # Temp file in the same directory so os.replace is a same-filesystem rename.
    fd, tmp_name = tempfile.mkstemp(
        prefix=f".{dest.name}.", suffix=".tmp", dir=str(dest.parent)
    )
    tmp_path = Path(tmp_name)
    try:
        with os.fdopen(fd, "wb") as fh:
            fh.write(raw)
            fh.flush()
            os.fsync(fh.fileno())
        os.replace(tmp_path, dest)
        _fsync_dir(dest.parent)
    except BaseException:
        # Best-effort cleanup of the temp file on any failure.
        try:
            tmp_path.unlink()
        except FileNotFoundError:
            pass
        raise
    return dest

atomic_graph_save

atomic_graph_save(graph: Any, path: Union[str, PathLike]) -> Path

Persist a KGLite graph to path atomically.

KGLite writes the cache file itself (it can't be handed pre-serialized bytes, so :func:atomic_write doesn't apply), so the graph is saved to a uniquely-named temp file in the same directory and then os.replace-d into place. The unique name (pid + uuid) matters because more than one process can share a single cache file -- e.g. the memory MCP server and the dashboard backend both persist memory.kgl -- and a fixed temp path would let two concurrent saves clobber each other into a torn file. os.replace is atomic on the same filesystem, so a concurrent reader never sees a half-written cache and the worst case is a harmless last-writer-wins (the cache is disposable and rebuilt from the source-of-truth files).

Returns the destination :class:~pathlib.Path.

Source code in memory/safety.py

def atomic_graph_save(graph: Any, path: Union[str, os.PathLike]) -> Path:
    """Persist a KGLite graph to ``path`` atomically.

    KGLite writes the cache file itself (it can't be handed pre-serialized
    bytes, so :func:`atomic_write` doesn't apply), so the graph is saved to a
    uniquely-named temp file in the same directory and then ``os.replace``-d into
    place. The unique name (pid + uuid) matters because more than one process can
    share a single cache file -- e.g. the memory MCP server and the dashboard
    backend both persist ``memory.kgl`` -- and a fixed temp path would let two
    concurrent saves clobber each other into a torn file. ``os.replace`` is atomic
    on the same filesystem, so a concurrent reader never sees a half-written cache
    and the worst case is a harmless last-writer-wins (the cache is disposable and
    rebuilt from the source-of-truth files).

    Returns the destination :class:`~pathlib.Path`.
    """
    dest = Path(path)
    dest.parent.mkdir(parents=True, exist_ok=True)
    tmp = dest.with_name(f".{dest.name}.{os.getpid()}.{uuid.uuid4().hex}.tmp")
    try:
        graph.save(str(tmp))
        os.replace(tmp, dest)
    except BaseException:
        try:
            tmp.unlink()
        except FileNotFoundError:
            pass
        raise
    return dest

file_lock

file_lock(path: Union[str, PathLike], *, timeout: float = 10.0, poll: float = 0.05, stale_after: float = 600.0) -> Iterator[None]

Best-effort cross-process advisory lock via an exclusive-create lockfile.

Uses os.open(..., O_CREAT | O_EXCL), which is atomic on local filesystems on both Windows and POSIX. A lockfile older than stale_after seconds is treated as abandoned (e.g. a crashed process) and reclaimed.

Lockfiles should live under a runtime directory (e.g. the gitignored .angelo/), not inside a cloud-synced tree, to avoid sync interference.

Raises :class:TimeoutError if the lock cannot be acquired within timeout.

Source code in memory/safety.py

@contextmanager
def file_lock(
    path: Union[str, os.PathLike],
    *,
    timeout: float = 10.0,
    poll: float = 0.05,
    stale_after: float = 600.0,
) -> Iterator[None]:
    """Best-effort cross-process advisory lock via an exclusive-create lockfile.

    Uses ``os.open(..., O_CREAT | O_EXCL)``, which is atomic on local filesystems
    on both Windows and POSIX. A lockfile older than ``stale_after`` seconds is
    treated as abandoned (e.g. a crashed process) and reclaimed.

    Lockfiles should live under a runtime directory (e.g. the gitignored
    ``.angelo/``), not inside a cloud-synced tree, to avoid sync interference.

    Raises :class:`TimeoutError` if the lock cannot be acquired within ``timeout``.
    """
    lock_path = Path(path)
    lock_path.parent.mkdir(parents=True, exist_ok=True)
    start = time.monotonic()

    while True:
        try:
            fd = os.open(str(lock_path), os.O_CREAT | os.O_EXCL | os.O_WRONLY)
            try:
                os.write(fd, f"{os.getpid()} {time.time()}".encode())
            finally:
                os.close(fd)
            break
        except (FileExistsError, PermissionError) as exc:
            # FileExistsError: the lock is held. PermissionError is only treated
            # as contention on Windows, where an O_EXCL create against a file
            # that is mid-deletion (the holder's unlink) returns
            # ERROR_ACCESS_DENIED rather than EEXIST. On POSIX (Linux/macOS) a
            # PermissionError is a genuine problem (e.g. an unwritable lock dir),
            # so surface it immediately instead of masking it as a timeout.
            if isinstance(exc, PermissionError) and not _IS_WINDOWS:
                raise
            # Reclaim a stale lock if its mtime is older than stale_after.
            try:
                age = time.time() - lock_path.stat().st_mtime
                if age > stale_after:
                    try:
                        lock_path.unlink()
                    except FileNotFoundError:
                        pass
                    continue
            except FileNotFoundError:
                # Lock vanished between the failed create and the stat; retry.
                continue
            if time.monotonic() - start >= timeout:
                raise TimeoutError(
                    f"could not acquire lock {lock_path!s} within {timeout}s"
                )
            time.sleep(poll)

    try:
        yield
    finally:
        try:
            lock_path.unlink()
        except FileNotFoundError:
            pass