fix(mosaic): close credential evidence gaps

2026-08-05 18:01:41 -05:00
parent d18e49e8f4
commit 26203bd92c
16 changed files with 373 additions and 85 deletions
@@ -0,0 +1,78 @@
+#!/usr/bin/env python3
+"""Fail-closed reader for one governed Mosaic credential envelope."""
+
+import hashlib
+import json
+import os
+import stat
+import sys
+
+MAX_BYTES = 64 * 1024
+EXPECTED_KEYS = {
+    "schemaVersion",
+    "identity",
+    "estate",
+    "host",
+    "providerLogin",
+    "tokenName",
+    "scopes",
+    "createdAt",
+    "tokenDigest",
+    "token",
+}
+
+
+def refuse(message: str) -> None:
+    print(f"credential envelope refused: {message}", file=sys.stderr)
+    raise SystemExit(1)
+
+
+if len(sys.argv) != 5:
+    refuse("expected path, identity, estate, and host")
+path, identity, estate, host = sys.argv[1:]
+if not estate:
+    refuse("explicit estate is required")
+parent = os.path.dirname(path)
+try:
+    parent_stat = os.stat(parent, follow_symlinks=False)
+except OSError:
+    refuse("credential directory unavailable")
+if not stat.S_ISDIR(parent_stat.st_mode) or stat.S_ISLNK(parent_stat.st_mode):
+    refuse("credential directory is not a real directory")
+if parent_stat.st_uid != os.getuid() or parent_stat.st_mode & 0o022:
+    refuse("credential directory owner or mode is unsafe")
+try:
+    fd = os.open(path, os.O_RDONLY | os.O_NOFOLLOW | os.O_CLOEXEC)
+except OSError:
+    refuse("credential file unavailable or symbolic")
+try:
+    file_stat = os.fstat(fd)
+    if not stat.S_ISREG(file_stat.st_mode):
+        refuse("credential is not a regular file")
+    if file_stat.st_uid != os.getuid() or file_stat.st_mode & 0o077:
+        refuse("credential owner or mode is unsafe")
+    content = os.read(fd, MAX_BYTES + 1)
+    if len(content) > MAX_BYTES:
+        refuse("credential exceeds size limit")
+finally:
+    os.close(fd)
+try:
+    value = json.loads(content)
+except (UnicodeDecodeError, json.JSONDecodeError):
+    refuse("credential is not valid JSON")
+if not isinstance(value, dict) or set(value) != EXPECTED_KEYS:
+    refuse("credential schema is not exact")
+if (
+    value.get("schemaVersion") != 1
+    or value.get("identity") != identity
+    or value.get("estate") != estate
+    or value.get("host") != host
+    or value.get("providerLogin") != identity
+):
+    refuse("credential binding does not match requested identity, estate, host, and principal")
+token = value.get("token")
+if not isinstance(token, str) or not token or any(ch.isspace() for ch in token):
+    refuse("credential token is invalid")
+if value.get("tokenDigest") != hashlib.sha256(token.encode()).hexdigest():
+    refuse("credential digest does not match token")
+sys.stdout.write(token + "\n")