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
82 changes: 76 additions & 6 deletions src/omind/rules.py
Original file line number Diff line number Diff line change
Expand Up @@ -24,10 +24,27 @@
raised — a broken rule must never brick the guard. A note rule with the same
``id`` as a seed rule replaces it, so exceptions stay operator-editable.

v1 conditions are ``repo_visibility`` (via ``gh repo view``, cached one day,
Conditions are ``repo_visibility`` (via ``gh repo view``, cached one day,
**fail-open to UNKNOWN**: a rule conditioned on visibility does not fire when
visibility cannot be determined) and ``branch`` (the repo's checked-out
branch). ``except_repos`` matches the origin remote's repository name.
visibility cannot be determined), ``branch`` (the repo's checked-out branch),
and ``repo_has_commits`` (whether the *remote* holds any commit yet).
``except_repos`` matches the origin remote's repository name.

``repo_has_commits`` exists so a deny can be narrowed to repos that actually
have history — an initial commit into an empty repo has nothing to open a pull
request against, so the branch+PR ceremony cannot apply to it::

when:
repo_visibility: public
branch: [main, master]
repo_has_commits: true # only deny once there IS history to protect

It reads the REMOTE, not the local checkout: you must commit locally before you
can push, so a local check would never see an empty repo. Unlike visibility this
condition **fails safe** — when the remote cannot be reached the condition is
treated as satisfied, so an unreachable network can never silently hand out the
empty-repo exemption. Narrowing a deny is exactly where fail-open would be
wrong.
"""

from __future__ import annotations
Expand Down Expand Up @@ -69,13 +86,17 @@ class NoteRule:
message: str
when_visibility: str = ""
when_branch: tuple[str, ...] = ()
when_has_commits: bool | None = None
except_repos: tuple[str, ...] = ()
note: str = "(seed)"
invalid: str = "" # non-empty on a skipped block: the reason, for `rules list`

def conditioned_on_visibility(self) -> bool:
return bool(self.when_visibility)

def conditioned_on_has_commits(self) -> bool:
return self.when_has_commits is not None


SEED_NOTE_RULES = (
NoteRule(
Expand Down Expand Up @@ -128,6 +149,16 @@ def _parse_block(text: str, note: str) -> NoteRule:
excepts = [excepts]
excepts = tuple(str(r).strip() for r in excepts or [] if str(r).strip())
problems = []
has_commits: bool | None = None
if "repo_has_commits" in when:
raw = when.get("repo_has_commits")
if isinstance(raw, bool):
has_commits = raw
else:
# A bare `repo_has_commits: yes` parses as bool in YAML, but a typo
# like `repo_has_commits: "true"` would silently become truthy, so
# anything non-boolean is rejected loudly rather than guessed at.
problems.append("repo_has_commits must be true or false")
if not rule_id:
problems.append("missing id")
if not tool:
Expand All @@ -150,6 +181,7 @@ def _parse_block(text: str, note: str) -> NoteRule:
message=message,
when_visibility=visibility,
when_branch=branches,
when_has_commits=has_commits,
except_repos=excepts,
note=note,
)
Expand Down Expand Up @@ -320,6 +352,34 @@ def _repo_name(repo: Path) -> str:
return name[:-4] if name.endswith(".git") else name


def _remote_has_commits(repo: Path) -> bool | None:
"""Whether ``origin`` holds any commit yet. ``None`` when undeterminable.

Deliberately reads the remote rather than the local checkout: a push is
always preceded by a local commit, so a local probe would report "has
commits" every time and the empty-repo case could never be detected.

Not cached. A repo goes from empty to non-empty exactly once, and that
single transition is the whole point of the condition — a stale cache would
keep granting the exemption after the first commit landed.
"""
try:
proc = subprocess.run(
["git", "-C", str(repo), "ls-remote", "--heads", "origin"],
capture_output=True,
text=True,
timeout=10,
)
except (OSError, subprocess.SubprocessError):
_breadcrumb(f"rules_has_commits({repo})", "ls-remote failed")
return None
if proc.returncode != 0:
# No origin, no network, or no auth. Unknown, not "empty".
_breadcrumb(f"rules_has_commits({repo})", "ls-remote returned non-zero")
return None
return bool(proc.stdout.strip())


def _repo_branch(repo: Path) -> str:
try:
proc = subprocess.run(
Expand Down Expand Up @@ -405,7 +465,10 @@ def evaluate(
if not fnmatch.fnmatch(target, rule.match):
continue
repo_scoped = (
rule.conditioned_on_visibility() or rule.when_branch or rule.except_repos
rule.conditioned_on_visibility()
or rule.when_branch
or rule.except_repos
or rule.conditioned_on_has_commits()
)
if repo_scoped:
if repo is None:
Expand All @@ -417,6 +480,12 @@ def evaluate(
branches = pushed if pushed is not None else [_repo_branch(repo)]
if not any(branch in rule.when_branch for branch in branches):
continue
if rule.conditioned_on_has_commits():
has_commits = _remote_has_commits(repo)
# Fail SAFE: an undeterminable remote must never widen an
# exemption, so unknown is treated as satisfying the condition.
if has_commits is not None and has_commits != rule.when_has_commits:
continue
if rule.conditioned_on_visibility():
visibility = _repo_visibility(repo)
if visibility == _VISIBILITY_UNKNOWN:
Expand Down Expand Up @@ -476,11 +545,12 @@ def format_rules(omi_dir: Path | str) -> str:
conditions.append(f"visibility={rule.when_visibility}")
if rule.when_branch:
conditions.append(f"branch in {list(rule.when_branch)}")
if rule.conditioned_on_has_commits():
conditions.append(f"has_commits={str(rule.when_has_commits).lower()}")
if rule.except_repos:
conditions.append(f"except {list(rule.except_repos)}")
cond = f" when {', '.join(conditions)}" if conditions else ""
lines.append(
f"[{rule.action}] {rule.id}: {rule.tool} {rule.match!r}{cond} "
f"(from {rule.note})"
f"[{rule.action}] {rule.id}: {rule.tool} {rule.match!r}{cond} (from {rule.note})"
)
return "\n".join(lines) if lines else "(no rules)"
148 changes: 143 additions & 5 deletions tests/test_rules.py
Original file line number Diff line number Diff line change
Expand Up @@ -89,8 +89,7 @@ def repo(tmp_path: Path) -> Path:
repo.mkdir()
subprocess.run(["git", "init", "-q", "-b", "main", str(repo)], check=True)
subprocess.run(
["git", "-C", str(repo), "remote", "add", "origin",
"https://github.com/o/some-repo.git"],
["git", "-C", str(repo), "remote", "add", "origin", "https://github.com/o/some-repo.git"],
check=True,
)
return repo
Expand Down Expand Up @@ -157,9 +156,7 @@ def test_repo_scoped_rule_needs_a_repo(tmp_path: Path) -> None:
assert rules.evaluate(_action("git push origin main"), omi, None) is None


def test_guard_check_action_denies_via_note_rule(
tmp_path: Path, repo: Path, monkeypatch
) -> None:
def test_guard_check_action_denies_via_note_rule(tmp_path: Path, repo: Path, monkeypatch) -> None:
omi = tmp_path / "OMI"
_note_with_rule(omi)
monkeypatch.setattr(rules, "_repo_visibility", lambda r, **k: "public")
Expand Down Expand Up @@ -311,3 +308,144 @@ def test_remote_urls_pulled_out_of_git_remote_v() -> None:
"ssh://hermes/srv/git/omi.git",
]
assert rules._remote_urls("") == []


# --- repo_has_commits (empty-repo exemption) --------------------------------
#
# An initial commit into an empty repo has nothing to open a PR against, so the
# branch+PR deny must be narrowable to repos that actually have history.


def _note_with_has_commits(omi: Path, value: str) -> None:
omi.mkdir(parents=True, exist_ok=True)
(omi / "Guard Rules.md").write_text(
"```omind-rule\n"
"id: no-direct-push-public-main\n"
"tool: Bash\n"
'match: "*git push*"\n'
"when:\n"
" repo_visibility: public\n"
" branch: [main, master]\n"
f" repo_has_commits: {value}\n"
"action: deny\n"
'message: "Public repo: branch + PR required."\n'
"```\n",
encoding="utf-8",
)


def test_has_commits_parsed_as_bool(tmp_path: Path) -> None:
omi = tmp_path / "OMI"
_note_with_has_commits(omi, "true")
rule = rules.load_rules(omi)[0]
assert rule.when_has_commits is True
assert rule.conditioned_on_has_commits()

_note_with_has_commits(omi, "false")
rules._file_cache.clear()
assert rules.load_rules(omi)[0].when_has_commits is False


def test_has_commits_absent_means_unconditioned(tmp_path: Path) -> None:
omi = tmp_path / "OMI"
_note_with_rule(omi)
rule = rules.load_rules(omi)[0]
assert rule.when_has_commits is None
assert not rule.conditioned_on_has_commits()


def test_has_commits_rejects_non_boolean(tmp_path: Path) -> None:
"""A quoted "true" is a string, not a bool. Guessing at it would silently
flip a deny, so the block is skipped and reported instead."""
omi = tmp_path / "OMI"
_note_with_has_commits(omi, '"yes please"')
everything = rules.load_rules(omi, include_invalid=True)
assert any("repo_has_commits" in r.invalid for r in everything)
# The bad block does not replace the seed rule -- the guard stays armed
# rather than silently vanishing, which is the failure mode that makes a
# typo'd exception so dangerous.
live = rules.load_rules(omi)
assert [r.id for r in live] == ["no-direct-push-public-main"]
assert live[0].note == "(seed)"
assert live[0].when_has_commits is None


def test_deny_fires_when_remote_has_commits(tmp_path: Path, repo: Path, monkeypatch) -> None:
omi = tmp_path / "OMI"
_note_with_has_commits(omi, "true")
monkeypatch.setattr(rules, "_repo_visibility", lambda r, **k: "public")
monkeypatch.setattr(rules, "_repo_branch", lambda r: "main")
monkeypatch.setattr(rules, "_remote_has_commits", lambda r: True)
hit = rules.evaluate(_action("git push origin main"), omi, repo)
assert hit is not None and hit.outcome == "deny"


def test_deny_skipped_when_remote_is_empty(tmp_path: Path, repo: Path, monkeypatch) -> None:
"""The whole point: an empty remote is exempt from branch+PR."""
omi = tmp_path / "OMI"
_note_with_has_commits(omi, "true")
monkeypatch.setattr(rules, "_repo_visibility", lambda r, **k: "public")
monkeypatch.setattr(rules, "_repo_branch", lambda r: "main")
monkeypatch.setattr(rules, "_remote_has_commits", lambda r: False)
assert rules.evaluate(_action("git push origin main"), omi, repo) is None


def test_unknown_remote_fails_safe_and_still_denies(
tmp_path: Path, repo: Path, monkeypatch
) -> None:
"""Opposite of visibility's fail-open. An unreachable remote must never
hand out the empty-repo exemption, so unknown keeps the deny."""
omi = tmp_path / "OMI"
_note_with_has_commits(omi, "true")
monkeypatch.setattr(rules, "_repo_visibility", lambda r, **k: "public")
monkeypatch.setattr(rules, "_repo_branch", lambda r: "main")
monkeypatch.setattr(rules, "_remote_has_commits", lambda r: None)
hit = rules.evaluate(_action("git push origin main"), omi, repo)
assert hit is not None and hit.outcome == "deny"


def test_remote_has_commits_probe_reads_the_remote(tmp_path: Path) -> None:
"""Empty remote -> False; after a commit is pushed -> True. Uses real git
against a local bare remote, so the probe itself is exercised."""
bare = tmp_path / "bare.git"
subprocess.run(["git", "init", "-q", "--bare", str(bare)], check=True)
work = tmp_path / "work"
work.mkdir()
subprocess.run(["git", "init", "-q", "-b", "main", str(work)], check=True)
subprocess.run(["git", "-C", str(work), "remote", "add", "origin", str(bare)], check=True)
# Local commit exists, remote is still empty -- the exact case that makes a
# local-only probe useless.
(work / "f.txt").write_text("x", encoding="utf-8")
subprocess.run(["git", "-C", str(work), "add", "-A"], check=True)
subprocess.run(
[
"git",
"-C",
str(work),
"-c",
"user.email=t@t",
"-c",
"user.name=t",
"commit",
"-qm",
"initial",
],
check=True,
)
assert rules._remote_has_commits(work) is False

subprocess.run(["git", "-C", str(work), "push", "-q", "origin", "main"], check=True)
assert rules._remote_has_commits(work) is True


def test_remote_has_commits_unknown_without_origin(tmp_path: Path) -> None:
solo = tmp_path / "solo"
solo.mkdir()
subprocess.run(["git", "init", "-q", "-b", "main", str(solo)], check=True)
assert rules._remote_has_commits(solo) is None


def test_format_rules_shows_has_commits(tmp_path: Path) -> None:
omi = tmp_path / "OMI"
_note_with_has_commits(omi, "true")
assert "has_commits=true" in rules.format_rules(omi)