content_gate.py

content_gate.py

1. #!/usr/bin/env python3
2. """
3. content_gate.py: optional content-level gate, run on staging after gate.py.
5. gate.py walls off folders by path. This scans each staged file and removes it (or,
6. in redact mode, replaces the tripped lines) when its frontmatter or a line trips a
7. rule, so a private line filed in an ordinary note is still caught before the
8. distiller reads it. It fails closed: a file it cannot read or decode is removed.
10. Env: BM_STAGING_DIR (the staging directory gate.py writes, .../cowork-staging/staging)
11. BM_GATE_MODE ("remove", the default, or "redact")
12. """
14. import json
15. import os
16. import re
17. import sys
18. from datetime import datetime, timezone
19. from pathlib import Path
21. STAGING_DIR = os.environ.get(
22. "BM_STAGING_DIR",
23. "/Users/CHANGEME/cowork-staging/staging",
24. )
26. # Mode: "remove" pulls a tripped file out of staging entirely.
27. # "redact" keeps the file but replaces tripped lines with a marker.
28. GATE_MODE = os.environ.get("BM_GATE_MODE", "remove")
29. MODES = ("remove", "redact")
31. # The same suffixes gate.py stages. Scanning only *.md would let a staged .txt
32. # through unchecked.
33. SCAN_SUFFIXES = {".md", ".markdown", ".txt"}
35. # Rules. A file whose frontmatter has any FRONTMATTER_BLOCK key: value pair is
36. # gated whole, so adding "private: true" to a note keeps it out wherever it lives.
37. # A LINE_PATTERNS match gates the whole file in remove mode and only the matching
38. # line in redact mode.
39. FRONTMATTER_BLOCK = [
40. ("private", "true"),
41. ("gate", "block"),
42. ("visibility", "private"),
43. ]
45. LINE_PATTERNS = [
46. # Obsidian-style inline private marker you can drop on any line:
47. # %% bm:private %% ... whole file gated in remove mode
48. re.compile(r"%%\s*bm:private\s*%%", re.IGNORECASE),
49. re.compile(r"\b\d{3}-\d{2}-\d{4}\b"), # SSN-shaped
50. ]
52. REDACTION_MARKER = "[[bm-gated]]"
55. def parse_frontmatter(text: str) -> dict:
56. """Minimal YAML-frontmatter key:value reader. No external deps."""
57. if not text.startswith("---"):
58. return {}
59. end = text.find("\n---", 3)
60. if end == -1:
61. return {}
62. block = text[3:end]
63. fm = {}
64. for line in block.splitlines():
65. if ":" in line:
66. k, _, v = line.partition(":")
67. fm[k.strip().lower()] = v.strip().strip("\"'").lower()
68. return fm
71. def file_trips_frontmatter(fm: dict) -> bool:
72. for key, val in FRONTMATTER_BLOCK:
73. if fm.get(key) == val:
74. return True
75. return False
78. def main() -> int:
79. staging = Path(STAGING_DIR).expanduser()
80. if "CHANGEME" in str(staging):
81. print("content_gate: refusing to run with placeholder staging path.", file=sys.stderr)
82. return 2
83. if not staging.is_dir():
84. print(f"content_gate: staging dir not found: {staging}", file=sys.stderr)
85. return 2
86. # An unknown mode must not quietly fall through to one of the others.
87. if GATE_MODE not in MODES:
88. print(f"content_gate: unknown BM_GATE_MODE {GATE_MODE!r} (use remove or redact).",
89. file=sys.stderr)
90. return 2
92. removed, redacted, passed = [], [], 0
94. for path in sorted(p for p in staging.iterdir()
95. if p.is_file() and p.suffix.lower() in SCAN_SUFFIXES):
96. try:
97. text = path.read_text(encoding="utf-8")
98. except (OSError, UnicodeDecodeError):
99. # Fail closed: a file we cannot read or decode gets pulled. A decode
100. # error is not an OSError, and letting it escape would crash the run
101. # with the rest of staging unscanned.
102. path.unlink(missing_ok=True)
103. removed.append(path.name + " (unreadable, fail-closed)")
104. continue
106. fm = parse_frontmatter(text)
107. tripped_whole = file_trips_frontmatter(fm)
109. line_hits = []
110. for i, line in enumerate(text.splitlines()):
111. if any(pat.search(line) for pat in LINE_PATTERNS):
112. line_hits.append(i)
114. if GATE_MODE == "remove":
115. if tripped_whole or line_hits:
116. path.unlink(missing_ok=True)
117. removed.append(path.name)
118. else:
119. passed += 1
120. else: # redact
121. if tripped_whole:
122. path.unlink(missing_ok=True)
123. removed.append(path.name + " (frontmatter-blocked)")
124. elif line_hits:
125. lines = text.splitlines()
126. for i in line_hits:
127. lines[i] = REDACTION_MARKER
128. path.write_text("\n".join(lines), encoding="utf-8")
129. redacted.append(path.name)
130. else:
131. passed += 1
133. report = {
134. "run_utc": datetime.now(timezone.utc).isoformat(),
135. "mode": GATE_MODE,
136. "passed": passed,
137. "removed": removed,
138. "redacted": redacted,
139. }
140. (staging / "gate-report.json").write_text(
141. json.dumps(report, indent=2), encoding="utf-8"
142. )
143. print(
144. f"content_gate: mode={GATE_MODE} passed={passed} "
145. f"removed={len(removed)} redacted={len(redacted)}"
146. )
147. return 0
150. if __name__ == "__main__":
151. raise SystemExit(main())