81 lines
2.6 KiB
Python
81 lines
2.6 KiB
Python
#!/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) != 6:
|
|
refuse("expected governed root, path, identity, estate, and host")
|
|
root, path, identity, estate, host = sys.argv[1:]
|
|
if os.path.abspath(os.path.dirname(path)) != os.path.abspath(root):
|
|
refuse("credential is not a direct child of the governed root")
|
|
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")
|