Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
73 changes: 73 additions & 0 deletions src/omind/scope_guard.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,73 @@
# SPDX-License-Identifier: Apache-2.0
# Copyright 2026 Aaron K. Clark
"""Scoped-write interlock (2026-08-27 roundtable, item #5).

An operational guard against *accidentally* writing a note into the wrong
project's scope — a fat-fingered cross-project write becomes a loud, remediable
error instead of a silent one.

**This is not a security boundary.** It is deliberately bypassable: unset
``OMIND_SCOPE``, edit the Markdown file directly, or run any process that does
not pass through the guarded write tools, and the check does not apply. The
panel adopted the reframe unanimously — the doc must say this plainly rather
than imply an enforcement it cannot deliver.

Enforcement is opt-in on BOTH sides and fails open otherwise:

* an undeclared ``OMIND_SCOPE`` never blocks anything (fail-open — the common
case of running with no scope set stays frictionless);
* an unscoped note is always writable (the same asymmetry retrieval uses:
unscoped means global);
* only when the process declares a scope AND the note declares a *different*
one is the write denied — with ``OMIND_SCOPE_MODE=warn`` as the escape hatch
that downgrades the deny to a returned warning.
"""

from __future__ import annotations

import os
from collections.abc import Mapping

SCOPE_ENV = "OMIND_SCOPE"
MODE_ENV = "OMIND_SCOPE_MODE"


def _clean(value: object) -> str:
"""Normalise a scope label exactly as ``store._clean_scope`` does."""
return str(value or "").strip().lstrip("#").strip().lower()


class ScopeViolationError(ValueError):
"""An out-of-scope write refused in deny mode.

Subclasses ``ValueError`` so the MCP layer renders it as a tool error the
same way the other write-time validations do.
"""


def process_scope(env: Mapping[str, str] | None = None) -> str:
"""The scope this process declares, or ``""`` when unset (fail-open)."""
return _clean((env if env is not None else os.environ).get(SCOPE_ENV, ""))


def check_write(note_scope: object, *, env: Mapping[str, str] | None = None) -> str | None:
"""Guard one note write.

Returns ``None`` to allow, a warning string when ``OMIND_SCOPE_MODE=warn``
downgrades an out-of-scope write, and raises :class:`ScopeViolationError` to deny.
"""
env = env if env is not None else os.environ
proc = _clean(env.get(SCOPE_ENV, ""))
note = _clean(note_scope)
if not proc or not note or proc == note:
return None
message = (
f"scope interlock: this process runs under {SCOPE_ENV}={proc!r}, but the "
f"note is scoped {note!r} — refusing the out-of-scope write. To proceed, "
f"do one of: set the note's scope to {proc!r}; unset {SCOPE_ENV}; or set "
f"{MODE_ENV}=warn to downgrade this guard to a warning. (This interlock "
"guards against accidents; it is not a security boundary.)"
)
if _clean(env.get(MODE_ENV, "")) == "warn":
return message
raise ScopeViolationError(message)
23 changes: 22 additions & 1 deletion src/omind/server.py
Original file line number Diff line number Diff line change
Expand Up @@ -31,7 +31,7 @@
from mcp.server.mcpserver.exceptions import ToolError
from mcp.shared.message import SessionMessage

from omind import graph, searchindex
from omind import graph, scope_guard, searchindex
from omind.help_system import render_help
from omind.recall import DEFAULT_RECALL_CHARS, compact_recall
from omind.store import (
Expand All @@ -41,6 +41,7 @@
NoteFields,
OmiStore,
_clean_agent,
_clean_scope,
parse_note,
)

Expand Down Expand Up @@ -380,7 +381,15 @@ def create_note(
action_items: list[str] | None = None,
references: list[str] | None = None,
agent: str = "",
scope: str = "",
) -> dict[str, object]:
# Scoped-write interlock (item #5): deny an out-of-scope write BEFORE it
# lands. Raises (rendered as a tool error) in deny mode; returns a
# warning to surface in warn mode; fail-open when either side is
# unscoped. Never reached by mesh/system writes, which call the store
# directly rather than through this tool.
note_scope = _clean_scope(scope)
scope_warning = scope_guard.check_write(note_scope)
fields = NoteFields(
title=title,
summary=summary,
Expand All @@ -395,9 +404,12 @@ def create_note(
connections=connections or [],
action_items=_parse_action_items(action_items or []),
references=references or [],
scope=note_scope,
)
filename = store.create_note(fields)
result: dict[str, object] = {"filename": filename, "agent": fields.agent}
if scope_warning:
result["scope_warning"] = scope_warning
# Write-time near-duplicate warning (2026-08-27 roundtable: ADOPT —
# advisory, fail-open, never blocks the write; hint field DROPPED for
# v1 because cosine cannot distinguish "replaces" from "disagrees";
Expand Down Expand Up @@ -458,6 +470,7 @@ def edit_note(
action_items: list[str] | None = None,
references: list[str] | None = None,
agent: str | None = None,
scope: str | None = None,
expected_version: str | None = None,
) -> dict[str, str]:
fields = store.read_fields(name)
Expand Down Expand Up @@ -489,8 +502,16 @@ def edit_note(
# Omitted keeps the current writer; an explicit value re-attributes
# (e.g. a takeover). Resolution: arg > OMIND_AGENT env.
fields.agent = _resolve_agent(agent)
if scope is not None:
fields.scope = _clean_scope(scope)
# Scoped-write interlock (item #5): guard the effective post-edit scope
# before the write lands. An edit that leaves scope untouched is still
# guarded against the note's existing scope.
scope_warning = scope_guard.check_write(fields.scope)
filename = store.update_note(name, fields, expected_version=expected_version)
result: dict[str, str] = {"filename": filename, "version": store.note_version(name)}
if scope_warning:
result["scope_warning"] = scope_warning
if expected_version is None:
# Make the silent-last-write-wins case visible to the caller
# (2026-08-27 review): without the token the write was not checked
Expand Down
20 changes: 15 additions & 5 deletions src/omind/store.py
Original file line number Diff line number Diff line change
Expand Up @@ -205,11 +205,21 @@ class NoteFields:
#: a full copy, so every agent on every machine sees every note — fine for
#: precision when the vault is small, less so as it grows.
#:
#: NOT A SECURITY BOUNDARY, and must never be described as one. The note is
#: still plain Markdown on disk, still replicated to every peer, still
#: readable by anything with the file. This narrows what RETRIEVAL returns,
#: nothing more. Absent (the default, and every note ever written) means
#: unscoped, which stays visible to every query.
#: Two roles (2026-08-27 roundtable, item #5): it narrows what RETRIEVAL
#: returns, and it is a guard-enforced operational INTERLOCK against
#: accidents — when a process declares ``OMIND_SCOPE`` and a note declares a
#: different scope, the write tools refuse the write (``OMIND_SCOPE_MODE=warn``
#: downgrades that to a warning). Its job is to turn a fat-fingered
#: cross-project write into a loud error, nothing more.
#:
#: NOT A SECURITY BOUNDARY, and must never be described as one. The interlock
#: is deliberately BYPASSABLE: unset ``OMIND_SCOPE``, edit the Markdown file
#: directly, or run any process that does not go through the guarded write
#: tools (mesh sync and other system writes are exempt by design), and it
#: does not apply. The note is still plain Markdown on disk, still replicated
#: to every peer, still readable by anything with the file. Absent (the
#: default, and every note ever written) means unscoped, which stays visible
#: to every query and is always writable.
scope: str = ""
#: Self-declared identity of the last writer ("Agent: hermes/pluto/dix"),
#: resolved by the caller (explicit arg > OMIND_AGENT env > empty). 2026-08-27
Expand Down
49 changes: 49 additions & 0 deletions tests/test_scope_guard.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,49 @@
# SPDX-License-Identifier: Apache-2.0
# Copyright 2026 Aaron K. Clark
"""Tests for the scoped-write interlock (item #5)."""

from __future__ import annotations

import pytest

from omind import scope_guard


def test_unset_process_scope_fails_open() -> None:
# The common case — no OMIND_SCOPE — never blocks, even a scoped note.
assert scope_guard.check_write("buzz", env={}) is None


def test_unscoped_note_is_always_writable() -> None:
assert scope_guard.check_write("", env={"OMIND_SCOPE": "buzz"}) is None


def test_matching_scope_is_allowed() -> None:
assert scope_guard.check_write("buzz", env={"OMIND_SCOPE": "buzz"}) is None


def test_scope_labels_are_normalised_before_comparison() -> None:
# "#Buzz" and "buzz" are the same scope, like store._clean_scope.
assert scope_guard.check_write("#Buzz", env={"OMIND_SCOPE": "buzz"}) is None


def test_out_of_scope_write_is_denied_by_default() -> None:
with pytest.raises(scope_guard.ScopeViolationError) as excinfo:
scope_guard.check_write("antigua", env={"OMIND_SCOPE": "buzz"})
message = str(excinfo.value)
assert "buzz" in message and "antigua" in message
assert "OMIND_SCOPE_MODE=warn" in message # remediation is spelled out
assert "not a security boundary" in message.lower()


def test_warn_mode_downgrades_deny_to_a_returned_warning() -> None:
warning = scope_guard.check_write(
"antigua", env={"OMIND_SCOPE": "buzz", "OMIND_SCOPE_MODE": "warn"}
)
assert warning is not None
assert "antigua" in warning


def test_process_scope_reads_and_normalises_the_env() -> None:
assert scope_guard.process_scope({"OMIND_SCOPE": "#Buzz "}) == "buzz"
assert scope_guard.process_scope({}) == ""
45 changes: 45 additions & 0 deletions tests/test_server.py
Original file line number Diff line number Diff line change
Expand Up @@ -562,3 +562,48 @@ def crash() -> None:
# mask and log with its traceback, exactly as the SDK intends.
with pytest.raises(OSError, match="disk on fire"):
crash()


# -- scoped-write interlock (item #5) --------------------------------------


def test_create_note_denies_an_out_of_scope_write(
server: MCPServer, omi_dir: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.setenv("OMIND_SCOPE", "buzz")
with pytest.raises(ToolError, match="scope interlock"):
call(server, "create-note", {"title": "Wrong", "summary": "s", "scope": "antigua"})
assert not (omi_dir / "Wrong.md").exists() # the deny prevented the write


def test_create_note_warn_mode_writes_but_flags_it(
server: MCPServer, omi_dir: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.setenv("OMIND_SCOPE", "buzz")
monkeypatch.setenv("OMIND_SCOPE_MODE", "warn")
result = call(server, "create-note", {"title": "Soft", "summary": "s", "scope": "antigua"})
assert result["filename"] == "Soft.md"
assert "antigua" in result["scope_warning"]
assert (omi_dir / "Soft.md").is_file()


def test_in_scope_and_unscoped_writes_are_allowed(
server: MCPServer, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.setenv("OMIND_SCOPE", "buzz")
matched = call(server, "create-note", {"title": "Right", "summary": "s", "scope": "buzz"})
assert "scope_warning" not in matched
# An unscoped note is global — always writable, even under a scoped process.
unscoped = call(server, "create-note", {"title": "Global", "summary": "s"})
assert "scope_warning" not in unscoped


def test_edit_is_guarded_against_the_notes_existing_scope(
server: MCPServer, monkeypatch: pytest.MonkeyPatch
) -> None:
# Created with no interlock (process unscoped), so the antigua note exists.
call(server, "create-note", {"title": "Owned", "summary": "s", "scope": "antigua"})
monkeypatch.setenv("OMIND_SCOPE", "buzz")
# An edit that never touches scope is still guarded against the note's scope.
with pytest.raises(ToolError, match="scope interlock"):
call(server, "edit-note", {"name": "Owned.md", "summary": "changed"})