gabriel / muse public
trust.py python
383 lines 13.8 KB
Raw
sha256:08c083095bcaffb4c43ce947668fac93bf5d261e13d321776170c4019dd1d77d fix: migration must never update identity.toml on a failed … Sonnet 5 minor ⚠ breaking 9 hours ago
1 """muse trust — manage trusted repository paths and pinned hub fingerprints.
2
3 Security model
4 --------------
5 Muse performs an ownership check on every discovered ``.muse/`` directory
6 (CVE-2022-24765 equivalent). If the ``.muse/`` directory is owned by a
7 different UID, Muse refuses to operate on it and raises
8 :class:`~muse.core.errors.UntrustedRepositoryError`.
9
10 For shared-filesystem environments (Docker bind-mounts, CI agents, multi-user
11 workstations), specific paths can be added to the trust list so Muse operates
12 normally without re-checking ownership.
13
14 Trust sources (in evaluation order)
15 ------------------------------------
16 1. ``MUSE_SAFE_DIRS`` environment variable — colon-separated absolute paths.
17 Useful for ephemeral CI containers.
18 2. ``~/.muse/config.toml`` ``[security] safe_dirs`` — persistent trust list.
19 Managed by ``muse trust add`` / ``muse trust remove``.
20
21 Hub fingerprint pinning (TOFU)
22 -------------------------------
23 ``muse trust hub-list`` and ``muse trust hub-reset`` manage the TOFU
24 (Trust On First Use) fingerprint store at ``~/.muse/hub_trust.toml``.
25
26 JSON output contract
27 --------------------
28 Every ``--json`` response includes ``duration_ms`` (float, milliseconds) and
29 ``exit_code`` (int). Errors are emitted to **stdout** (not stderr) in JSON
30 mode so agents can always parse a structured response::
31
32 {"error": "<key>", "message": "<description>", "duration_ms": 0.3, "exit_code": 3}
33
34 Exit codes::
35
36 0 — success
37 3 — I/O error reading or writing the trust store
38 """
39
40 import argparse
41 import json as _json
42 import sys
43 from typing import TYPE_CHECKING, TypedDict
44
45 from muse.core.envelope import EnvelopeJson, make_envelope
46 from muse.core.timing import start_timer
47
48 if TYPE_CHECKING:
49 from muse.core.hub_trust import HubTrustStore
50
51 class _TrustAddJson(EnvelopeJson):
52 added: bool
53 path: str
54
55 class _TrustRemoveJson(EnvelopeJson):
56 removed: bool
57 path: str
58
59 class _TrustListJson(EnvelopeJson):
60 trusted_dirs: list[str]
61 count: int
62
63 class _TrustHubListJson(EnvelopeJson):
64 fingerprints: HubTrustStore
65 count: int
66
67 class _TrustHubResetJson(EnvelopeJson):
68 reset: bool
69 hostname: str
70
71 def _add_json_flag(parser: argparse.ArgumentParser) -> None:
72 parser.add_argument(
73 "--json", "-j",
74 action="store_true",
75 default=False,
76 dest="json_out",
77 help="Emit machine-readable JSON instead of human-readable text.",
78 )
79 parser.set_defaults(json_out=False)
80
81 def register(subparsers: argparse._SubParsersAction[argparse.ArgumentParser]) -> None:
82 """Register the ``muse trust`` namespace and its subcommands."""
83 trust_parser = subparsers.add_parser(
84 "trust",
85 help="Manage trusted repository paths and pinned hub fingerprints.",
86 description=__doc__,
87 formatter_class=argparse.RawDescriptionHelpFormatter,
88 )
89 trust_subs = trust_parser.add_subparsers(dest="trust_command", metavar="TRUST_COMMAND")
90 trust_subs.required = True
91
92 # muse trust add <path>
93 add_p = trust_subs.add_parser(
94 "add",
95 help="Add a path to the trusted directory list (~/.muse/config.toml).",
96 description=(
97 "Add PATH to the [security] safe_dirs list in ~/.muse/config.toml.\n"
98 "Muse will skip the ownership check for this repository.\n\n"
99 "Example:\n"
100 " muse trust add /shared/workspaces/myrepo"
101 ),
102 formatter_class=argparse.RawDescriptionHelpFormatter,
103 )
104 add_p.add_argument("path", help="Repository path to trust (absolute or relative).")
105 _add_json_flag(add_p)
106 add_p.set_defaults(func=_run_add)
107
108 # muse trust remove <path>
109 remove_p = trust_subs.add_parser(
110 "remove",
111 help="Remove a path from the trusted directory list.",
112 description=(
113 "Remove PATH from the [security] safe_dirs list in ~/.muse/config.toml."
114 ),
115 formatter_class=argparse.RawDescriptionHelpFormatter,
116 )
117 remove_p.add_argument("path", help="Repository path to remove from the trust list.")
118 _add_json_flag(remove_p)
119 remove_p.set_defaults(func=_run_remove)
120
121 # muse trust list
122 list_p = trust_subs.add_parser(
123 "list",
124 help="Print all currently trusted paths.",
125 description="Print all paths in the [security] safe_dirs list.",
126 )
127 _add_json_flag(list_p)
128 list_p.set_defaults(func=_run_list)
129
130 # muse trust hub-list
131 hub_list_p = trust_subs.add_parser(
132 "hub-list",
133 help="List all pinned hub fingerprints (TOFU store).",
134 description=(
135 "Print all hub fingerprints stored in ~/.muse/hub_trust.toml.\n"
136 "Each entry shows: hostname, SHA-256 fingerprint, first seen, verified count."
137 ),
138 formatter_class=argparse.RawDescriptionHelpFormatter,
139 )
140 _add_json_flag(hub_list_p)
141 hub_list_p.set_defaults(func=_run_hub_list)
142
143 # muse trust hub-reset <hostname>
144 hub_reset_p = trust_subs.add_parser(
145 "hub-reset",
146 help="Remove a pinned hub fingerprint so the next connection re-pins (TOFU reset).",
147 description=(
148 "Remove the stored fingerprint for HOSTNAME from ~/.muse/hub_trust.toml.\n"
149 "The next connection to this hub will re-pin with TOFU.\n\n"
150 "Use this after a legitimate certificate rotation to acknowledge the change."
151 ),
152 formatter_class=argparse.RawDescriptionHelpFormatter,
153 )
154 hub_reset_p.add_argument("hostname", help="Hub hostname (e.g. 'musehub.ai' or 'localhost:1337').")
155 _add_json_flag(hub_reset_p)
156 hub_reset_p.set_defaults(func=_run_hub_reset)
157
158 trust_parser.set_defaults(func=lambda args: trust_parser.print_help() or sys.exit(0))
159
160 def _run_add(args: argparse.Namespace) -> None:
161 """Add a path to the global trusted-directory list.
162
163 Normalises *path* to an absolute path before storing. Adding the same
164 path twice is idempotent — no duplicate entry is created.
165
166 Agent quickstart::
167
168 muse trust add /shared/workspaces/myrepo --json
169 muse trust add . --json
170
171 JSON fields::
172
173 added true if the path was newly added (false if already trusted).
174 path Absolute path that was added.
175 muse_version Muse release that produced this output.
176 schema Envelope schema version (int).
177 exit_code 0 on success, 3 on I/O error.
178 duration_ms Wall-clock milliseconds for the command.
179 timestamp ISO-8601 UTC timestamp of command completion.
180 warnings List of non-fatal advisory messages.
181
182 Exit codes::
183
184 0 Path added (or already trusted).
185 3 I/O error reading or writing ~/.muse/config.toml.
186 """
187 elapsed = start_timer()
188 from muse.cli.config import add_global_safe_dir
189 import os
190 abs_path = os.path.abspath(args.path)
191 json_out: bool = getattr(args, "json_out", False)
192
193 def _emit_error(msg: str, code: int, error_key: str = "error") -> None:
194 if json_out:
195 print(_json.dumps({**make_envelope(elapsed, exit_code=code), "error": error_key, "message": msg}))
196 else:
197 print(f"❌ {msg}", file=sys.stderr)
198 raise SystemExit(code)
199
200 try:
201 add_global_safe_dir(abs_path)
202 except OSError as exc:
203 _emit_error(str(exc), 3, "io_error")
204
205 if json_out:
206 print(_json.dumps(_TrustAddJson(**make_envelope(elapsed), added=True, path=abs_path)))
207 else:
208 print(f"Trusted path added: {abs_path}")
209 print(f"Muse will skip the ownership check for: {abs_path}")
210
211 def _run_remove(args: argparse.Namespace) -> None:
212 """Remove a path from the global trusted-directory list.
213
214 Normalises *path* to an absolute path before matching. Removal is
215 idempotent — removing a path that is not trusted exits 0 with ``removed=false``.
216
217 Agent quickstart::
218
219 muse trust remove /shared/workspaces/myrepo --json
220
221 JSON fields::
222
223 removed true if the path was found and removed; false if not trusted.
224 path Absolute path that was targeted.
225 muse_version Muse release that produced this output.
226 schema Envelope schema version (int).
227 exit_code 0 on success, 3 on I/O error.
228 duration_ms Wall-clock milliseconds for the command.
229 timestamp ISO-8601 UTC timestamp of command completion.
230 warnings List of non-fatal advisory messages.
231
232 Exit codes::
233
234 0 Completed (whether or not the path was in the list).
235 3 I/O error reading or writing ~/.muse/config.toml.
236 """
237 elapsed = start_timer()
238 from muse.cli.config import remove_global_safe_dir, get_global_safe_dirs
239 import os
240 abs_path = os.path.abspath(args.path)
241 json_out: bool = getattr(args, "json_out", False)
242
243 def _emit_error(msg: str, code: int, error_key: str = "error") -> None:
244 if json_out:
245 print(_json.dumps({**make_envelope(elapsed, exit_code=code), "error": error_key, "message": msg}))
246 else:
247 print(f"❌ {msg}", file=sys.stderr)
248 raise SystemExit(code)
249
250 try:
251 before = get_global_safe_dirs()
252 remove_global_safe_dir(abs_path)
253 after = get_global_safe_dirs()
254 except OSError as exc:
255 _emit_error(str(exc), 3, "io_error")
256
257 removed = abs_path in before and abs_path not in after
258 if json_out:
259 print(_json.dumps(_TrustRemoveJson(**make_envelope(elapsed), removed=removed, path=abs_path)))
260 elif removed:
261 print(f"Trusted path removed: {abs_path}")
262 else:
263 print(f"Path not in trust list: {abs_path}")
264
265 def _run_list(args: argparse.Namespace) -> None:
266 """List all trusted paths from ``~/.muse/config.toml``.
267
268 An empty list is a valid result — exits 0 with ``count=0``.
269
270 Agent quickstart::
271
272 muse trust list --json
273
274 JSON fields::
275
276 trusted_dirs List of absolute trusted-directory paths.
277 count Number of trusted paths.
278 muse_version Muse release that produced this output.
279 schema Envelope schema version (int).
280 exit_code Always 0.
281 duration_ms Wall-clock milliseconds for the command.
282 timestamp ISO-8601 UTC timestamp of command completion.
283 warnings List of non-fatal advisory messages.
284
285 Exit codes::
286
287 0 Success (empty list is still success).
288 """
289 elapsed = start_timer()
290 from muse.cli.config import get_global_safe_dirs
291 dirs = get_global_safe_dirs()
292 if getattr(args, "json_out", False):
293 print(_json.dumps(_TrustListJson(**make_envelope(elapsed), trusted_dirs=dirs, count=len(dirs))))
294 return
295 if not dirs:
296 print("No trusted directories configured.")
297 print("Add paths with: muse trust add <path>")
298 return
299 print("Trusted directories (~/.muse/config.toml):")
300 for d in dirs:
301 print(f" {d}")
302
303 def _run_hub_list(args: argparse.Namespace) -> None:
304 """List all pinned hub TLS fingerprints from ``~/.muse/hub_trust.toml``.
305
306 Each entry is keyed by hostname and includes the full ``HubTrustRecord``
307 (``hub_url``, ``fingerprint``, ``first_seen``, ``verified_count``).
308
309 Agent quickstart::
310
311 muse trust hub-list --json
312
313 JSON fields::
314
315 fingerprints Map of hostname → HubTrustRecord (hub_url, fingerprint, first_seen, verified_count).
316 count Number of pinned fingerprints.
317 muse_version Muse release that produced this output.
318 schema Envelope schema version (int).
319 exit_code Always 0.
320 duration_ms Wall-clock milliseconds for the command.
321 timestamp ISO-8601 UTC timestamp of command completion.
322 warnings List of non-fatal advisory messages.
323
324 Exit codes::
325
326 0 Success (empty map is still success).
327 """
328 elapsed = start_timer()
329 from muse.core.hub_trust import load_hub_trust_store
330 store = load_hub_trust_store()
331 if getattr(args, "json_out", False):
332 print(_json.dumps(_TrustHubListJson(**make_envelope(elapsed), fingerprints=dict(store), count=len(store))))
333 return
334 if not store:
335 print("No hub fingerprints pinned.")
336 print("Fingerprints are pinned automatically on first connection.")
337 return
338 print("Pinned hub fingerprints (~/.muse/hub_trust.toml):")
339 for hostname, record in sorted(store.items()):
340 print(
341 f" {hostname}\n"
342 f" fingerprint: {record['fingerprint']}\n"
343 f" first_seen: {record['first_seen']}\n"
344 f" verified_count: {record['verified_count']}"
345 )
346
347 def _run_hub_reset(args: argparse.Namespace) -> None:
348 """Remove a hub fingerprint so the next connection re-pins (TOFU reset).
349
350 Idempotent — resetting an unknown hostname exits 0 with ``reset=false``.
351 Use after a legitimate TLS certificate rotation to acknowledge the change.
352
353 Agent quickstart::
354
355 muse trust hub-reset musehub.ai --json
356 muse trust hub-reset localhost:1337 --json
357
358 JSON fields::
359
360 reset true if a fingerprint was found and removed; false if unknown hostname.
361 hostname Hub hostname that was targeted.
362 muse_version Muse release that produced this output.
363 schema Envelope schema version (int).
364 exit_code Always 0.
365 duration_ms Wall-clock milliseconds for the command.
366 timestamp ISO-8601 UTC timestamp of command completion.
367 warnings List of non-fatal advisory messages.
368
369 Exit codes::
370
371 0 Success (whether or not a fingerprint existed for the hostname).
372 """
373 elapsed = start_timer()
374 from muse.core.hub_trust import remove_hub_record
375 hostname = args.hostname
376 removed = remove_hub_record(hostname)
377 if getattr(args, "json_out", False):
378 print(_json.dumps(_TrustHubResetJson(**make_envelope(elapsed), reset=removed, hostname=hostname)))
379 elif removed:
380 print(f"Hub fingerprint reset for: {hostname}")
381 print("The next connection will re-pin with TOFU.")
382 else:
383 print(f"No pinned fingerprint found for: {hostname}")
File History 1 commit
sha256:08c083095bcaffb4c43ce947668fac93bf5d261e13d321776170c4019dd1d77d fix: migration must never update identity.toml on a failed … Sonnet 5 minor ⚠ 9 hours ago