|
1 | 1 | #!/usr/bin/env python3 |
2 | | -"""Safely install or update Codex Rig's managed global-instruction block.""" |
| 2 | +"""Safely install, update, or remove Codex Rig's managed global-instruction block.""" |
3 | 3 |
|
4 | 4 | from __future__ import annotations |
5 | 5 |
|
@@ -74,6 +74,38 @@ def merged_payload(existing: bytes, block: bytes) -> tuple[bytes, str]: |
74 | 74 | return updated, "already current" if updated == existing else "updated" |
75 | 75 |
|
76 | 76 |
|
| 77 | +def stripped_payload(existing: bytes) -> tuple[bytes, str]: |
| 78 | + """Remove one authenticated managed block, preserving every external byte.""" |
| 79 | + try: |
| 80 | + existing.decode("utf-8") |
| 81 | + except UnicodeDecodeError as error: |
| 82 | + raise UnsafeGlobalAgentsState("existing AGENTS.md is not UTF-8; refusing write") from error |
| 83 | + |
| 84 | + begin_count = existing.count(BEGIN_PREFIX) |
| 85 | + end_count = existing.count(END_MARKER.rstrip(b"\n")) |
| 86 | + if begin_count == 0 and end_count == 0: |
| 87 | + return existing, "absent" |
| 88 | + if begin_count != 1 or end_count != 1: |
| 89 | + raise UnsafeGlobalAgentsState("managed markers are malformed or duplicated; refusing write") |
| 90 | + |
| 91 | + begin_match = BEGIN_PATTERN.search(existing) |
| 92 | + if begin_match is None: |
| 93 | + raise UnsafeGlobalAgentsState("managed begin marker is malformed; refusing write") |
| 94 | + end_index = existing.find(END_MARKER, begin_match.end()) |
| 95 | + if end_index < 0: |
| 96 | + raise UnsafeGlobalAgentsState("managed end marker is malformed; refusing write") |
| 97 | + body = existing[begin_match.end() : end_index] |
| 98 | + if sha256(body) != begin_match.group(1).decode("ascii"): |
| 99 | + raise UnsafeGlobalAgentsState("managed block was modified; refusing write") |
| 100 | + |
| 101 | + block_end = end_index + len(END_MARKER) |
| 102 | + updated = existing[: begin_match.start()] + existing[block_end:] |
| 103 | + # collapse the single separator install prepended so removal leaves no doubled blank line |
| 104 | + if updated.endswith(b"\n\n") and existing[: begin_match.start()].endswith(b"\n\n"): |
| 105 | + updated = updated[:-1] |
| 106 | + return updated, "removed" |
| 107 | + |
| 108 | + |
77 | 109 | def backup_target(target: Path, codex_home: Path, payload: bytes) -> Path: |
78 | 110 | """Create and verify a unique backup before changing an existing target.""" |
79 | 111 | backup_root = codex_home / "backups" / "codex-rig" |
@@ -140,19 +172,50 @@ def install_global_agents(source: Path, codex_home: Path) -> tuple[str, Path, Pa |
140 | 172 | return action, target, backup |
141 | 173 |
|
142 | 174 |
|
| 175 | +def remove_global_agents(codex_home: Path) -> tuple[str, Path, Path | None]: |
| 176 | + """Strip Codex Rig's managed block from AGENTS.md without touching user content.""" |
| 177 | + target = codex_home / "AGENTS.md" |
| 178 | + if not target.exists(): |
| 179 | + return "absent", target, None |
| 180 | + if target.is_symlink(): |
| 181 | + raise UnsafeGlobalAgentsState(f"target is a symlink; refusing write: {target}") |
| 182 | + if not target.is_file(): |
| 183 | + raise UnsafeGlobalAgentsState(f"target is not an ordinary file; refusing write: {target}") |
| 184 | + |
| 185 | + existing = target.read_bytes() |
| 186 | + updated, action = stripped_payload(existing) |
| 187 | + if action == "absent": |
| 188 | + return "absent", target, None |
| 189 | + |
| 190 | + backup = backup_target(target, codex_home, existing) |
| 191 | + if updated.strip() == b"": |
| 192 | + target.unlink() # file held only our block — remove it entirely |
| 193 | + return "removed-file", target, backup |
| 194 | + mode = stat.S_IMODE(target.stat().st_mode) |
| 195 | + atomic_write(target, updated, mode, existing) |
| 196 | + return "removed-block", target, backup |
| 197 | + |
| 198 | + |
143 | 199 | def parse_args() -> argparse.Namespace: |
144 | 200 | """Parse explicit source and Codex-home paths.""" |
145 | 201 | parser = argparse.ArgumentParser(description=__doc__) |
146 | | - parser.add_argument("--source", type=Path, required=True, help="packaged assets/AGENTS.md template") |
| 202 | + parser.add_argument("--source", type=Path, help="packaged assets/AGENTS.md template (required unless --remove)") |
147 | 203 | parser.add_argument("--codex-home", type=Path, required=True, help="target Codex home") |
148 | | - return parser.parse_args() |
| 204 | + parser.add_argument("--remove", action="store_true", help="strip the managed block instead of installing it") |
| 205 | + args = parser.parse_args() |
| 206 | + if not args.remove and args.source is None: |
| 207 | + parser.error("--source is required unless --remove is given") |
| 208 | + return args |
149 | 209 |
|
150 | 210 |
|
151 | 211 | def main() -> int: |
152 | | - """Run one fail-closed global-instruction installation.""" |
| 212 | + """Run one fail-closed global-instruction installation or removal.""" |
153 | 213 | args = parse_args() |
154 | 214 | try: |
155 | | - action, target, backup = install_global_agents(args.source, args.codex_home) |
| 215 | + if args.remove: |
| 216 | + action, target, backup = remove_global_agents(args.codex_home) |
| 217 | + else: |
| 218 | + action, target, backup = install_global_agents(args.source, args.codex_home) |
156 | 219 | except UnsafeGlobalAgentsState as error: |
157 | 220 | print(f"global-agents-error: {error}", file=sys.stderr) |
158 | 221 | return 4 |
|
0 commit comments