diff --git a/src/omind/rules.py b/src/omind/rules.py index cf82c88..9039bb3 100644 --- a/src/omind/rules.py +++ b/src/omind/rules.py @@ -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 @@ -69,6 +86,7 @@ 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` @@ -76,6 +94,9 @@ class NoteRule: 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( @@ -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: @@ -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, ) @@ -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( @@ -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: @@ -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: @@ -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)" diff --git a/tests/test_rules.py b/tests/test_rules.py index 54cba8e..9695293 100644 --- a/tests/test_rules.py +++ b/tests/test_rules.py @@ -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 @@ -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") @@ -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)