95 lines
3.2 KiB
Python
95 lines
3.2 KiB
Python
"""Fail-closed secret-file access without value projection or logging."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import stat
|
|
from pathlib import Path
|
|
|
|
|
|
class SecretFileError(RuntimeError):
|
|
"""Stable non-secret error raised for an invalid runtime secret file."""
|
|
|
|
|
|
def validate_secret_file(
|
|
path: Path,
|
|
*,
|
|
root: Path = Path("/run/secrets"),
|
|
expected_uid: int | None = None,
|
|
expected_gid: int | None = None,
|
|
max_bytes: int = 8192,
|
|
) -> bytes:
|
|
"""Return bounded bytes only after containment, mode, owner, and content checks."""
|
|
|
|
try:
|
|
resolved_root = root.resolve(strict=True)
|
|
resolved = path.resolve(strict=True)
|
|
resolved.relative_to(resolved_root)
|
|
metadata = resolved.stat()
|
|
except (OSError, ValueError) as exc:
|
|
raise SecretFileError("required secret file is unavailable") from exc
|
|
if not stat.S_ISREG(metadata.st_mode):
|
|
raise SecretFileError("required secret path is not a regular file")
|
|
mode = stat.S_IMODE(metadata.st_mode)
|
|
if os.name != "nt" and (mode & 0o077 or mode & 0o111 or not mode & 0o400):
|
|
raise SecretFileError("required secret file permissions are unsafe")
|
|
if expected_uid is not None and metadata.st_uid != expected_uid:
|
|
raise SecretFileError("required secret file owner is invalid")
|
|
if expected_gid is not None and metadata.st_gid != expected_gid:
|
|
raise SecretFileError("required secret file group is invalid")
|
|
try:
|
|
payload = resolved.read_bytes()
|
|
except OSError as exc:
|
|
raise SecretFileError("required secret file is unreadable") from exc
|
|
payload = payload.rstrip(b"\r\n")
|
|
if (
|
|
not payload
|
|
or len(payload) > max_bytes
|
|
or b"\x00" in payload
|
|
or b"\r" in payload.rstrip(b"\r\n")
|
|
or b"\n" in payload.rstrip(b"\r\n")
|
|
):
|
|
raise SecretFileError("required secret file content is invalid")
|
|
return payload
|
|
|
|
|
|
def read_secret_file(
|
|
path: Path,
|
|
*,
|
|
root: Path = Path("/run/secrets"),
|
|
expected_uid: int | None = None,
|
|
expected_gid: int | None = None,
|
|
max_bytes: int = 8192,
|
|
) -> str:
|
|
"""Read a UTF-8 secret at its internal consumption boundary."""
|
|
|
|
payload = validate_secret_file(
|
|
path,
|
|
root=root,
|
|
expected_uid=expected_uid,
|
|
expected_gid=expected_gid,
|
|
max_bytes=max_bytes,
|
|
)
|
|
try:
|
|
return payload.decode("utf-8")
|
|
except UnicodeDecodeError as exc:
|
|
raise SecretFileError("required secret file encoding is invalid") from exc
|
|
|
|
|
|
def validate_required_secret_paths(raw: str) -> None:
|
|
"""Validate a comma-separated production startup contract without returning values."""
|
|
|
|
if not raw.strip():
|
|
return
|
|
expected_uid = (
|
|
int(os.environ["SECRET_EXPECTED_UID"]) if os.getenv("SECRET_EXPECTED_UID") else None
|
|
)
|
|
expected_gid = (
|
|
int(os.environ["SECRET_EXPECTED_GID"]) if os.getenv("SECRET_EXPECTED_GID") else None
|
|
)
|
|
for value in raw.split(","):
|
|
path = Path(value.strip())
|
|
if not path.is_absolute():
|
|
raise SecretFileError("required secret path must be absolute")
|
|
validate_secret_file(path, expected_uid=expected_uid, expected_gid=expected_gid)
|