311 lines
12 KiB
Python
311 lines
12 KiB
Python
"""Fail-closed policy used only by the task-scoped Action writer."""
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import json
|
|
import stat
|
|
import subprocess
|
|
import re
|
|
from pathlib import Path, PurePosixPath, PureWindowsPath
|
|
from urllib.parse import unquote
|
|
|
|
import isolated_writer_path_policy as path_policy
|
|
|
|
WRITE_ENDPOINTS = frozenset({"/api/commit-file"})
|
|
ALLOWED_ROOTS = (
|
|
"active-runtime",
|
|
"docs",
|
|
"patches",
|
|
"reports",
|
|
"scripts",
|
|
"test",
|
|
"tests",
|
|
)
|
|
DENIED_ROOTS = ("baseline", "config", "deploy", "deployments", "ops", "runtime", "services")
|
|
_TRUE = frozenset({"1", "true", "yes", "on"})
|
|
_FALSE = frozenset({"0", "false", "no", "off"})
|
|
_DENY_FLAGS = (
|
|
"AIOS_GITOPS_ALLOW_MAIN_WRITE",
|
|
"AIOS_GITOPS_ALLOW_TAG_WRITE",
|
|
"AIOS_GITOPS_ALLOW_BASELINE_WRITE",
|
|
"AIOS_GITOPS_ALLOW_REGISTRY_WRITE",
|
|
"AIOS_GITOPS_ALLOW_ROLLBACK",
|
|
"AIOS_GITOPS_ALLOW_DEPLOYMENT",
|
|
)
|
|
|
|
|
|
def enforcement_required() -> bool:
|
|
return os.getenv("AIOS_GITOPS_ISOLATED_WRITER_ENFORCEMENT", "") == "required"
|
|
|
|
|
|
def allowed_branch() -> str:
|
|
value = os.getenv("AIOS_GITOPS_ALLOWED_BRANCH", "").strip()
|
|
if not value:
|
|
raise PermissionError("isolated_writer_missing_allowed_branch")
|
|
return value
|
|
|
|
|
|
def allowed_base_oid() -> str:
|
|
value = os.getenv("AIOS_GITOPS_ALLOWED_BASE_OID", "").strip().lower()
|
|
if len(value) != 40 or any(ch not in "0123456789abcdef" for ch in value):
|
|
raise PermissionError("isolated_writer_invalid_allowed_base_oid")
|
|
return value
|
|
|
|
|
|
def _git(workspace: Path, *args: str) -> str:
|
|
result = subprocess.run(["git", *args], cwd=workspace, text=True, capture_output=True)
|
|
if result.returncode != 0:
|
|
raise PermissionError("isolated_writer_git_validation_failed")
|
|
return result.stdout.strip()
|
|
|
|
|
|
def require_service_invariants(workspace: Path, bare_repo: Path) -> None:
|
|
if not enforcement_required():
|
|
return
|
|
if os.getenv("AIOS_GITOPS_WRITES_ENABLED", "").strip().lower() not in _TRUE:
|
|
raise PermissionError("isolated_writer_write_gate_required")
|
|
if os.getenv("AIOS_GITOPS_RUNTIME_ROLE", "") != "isolated-writer":
|
|
raise PermissionError("isolated_writer_role_required")
|
|
if os.getenv("AIOS_GITOPS_WORKSPACE_ROLE", "") != "isolated-writer":
|
|
raise PermissionError("isolated_writer_workspace_role_required")
|
|
if os.getenv("AIOS_GITOPS_TOKEN_POLICY_VERSION_REQUIRED", "") != "2":
|
|
raise PermissionError("isolated_writer_token_policy_v2_required")
|
|
if not os.getenv("AIOS_GITOPS_SERVICE_AUDIENCE", "").strip():
|
|
raise PermissionError("isolated_writer_service_audience_required")
|
|
if not os.getenv("AIOS_GITOPS_TOKEN_REGISTRY_V2", "").strip():
|
|
raise PermissionError("isolated_writer_token_registry_required")
|
|
if not os.getenv("AIOS_GITOPS_WRITER_OPERATION_LEDGER", "").strip():
|
|
raise PermissionError("isolated_writer_operation_ledger_required")
|
|
path_policy.configured_path_policy()
|
|
for name in _DENY_FLAGS:
|
|
if os.getenv(name, "").strip().lower() not in _FALSE:
|
|
raise PermissionError("isolated_writer_forbidden_capability_flag")
|
|
if os.getenv("AIOS_GITOPS_REPOSITORY_IDENTITY", "") != "aios/aios":
|
|
raise PermissionError("isolated_writer_repository_identity_rejected")
|
|
configured_bare_value = os.getenv("AIOS_GITOPS_AUTHORITATIVE_GIT_DIR", "").strip()
|
|
if not configured_bare_value:
|
|
raise PermissionError("isolated_writer_authoritative_repo_rejected")
|
|
configured_bare = Path(configured_bare_value)
|
|
if configured_bare.resolve(strict=True) != bare_repo.resolve(strict=True):
|
|
raise PermissionError("isolated_writer_authoritative_repo_rejected")
|
|
writer_root_value = os.getenv("AIOS_GITOPS_APPROVED_WRITER_ROOT", "").strip()
|
|
if not writer_root_value:
|
|
raise PermissionError("isolated_writer_workspace_outside_approved_root")
|
|
writer_root = Path(writer_root_value).resolve(strict=True)
|
|
lexical_workspace = Path(os.path.abspath(str(workspace)))
|
|
canonical_workspace = lexical_workspace.resolve(strict=True)
|
|
if lexical_workspace != canonical_workspace:
|
|
raise PermissionError("isolated_writer_workspace_symlink_rejected")
|
|
try:
|
|
canonical_workspace.relative_to(writer_root)
|
|
except (ValueError, FileNotFoundError):
|
|
raise PermissionError("isolated_writer_workspace_outside_approved_root") from None
|
|
if _git(workspace, "rev-parse", "--abbrev-ref", "HEAD") != allowed_branch():
|
|
raise PermissionError("isolated_writer_current_branch_rejected")
|
|
approved_base = allowed_base_oid()
|
|
try:
|
|
_git(workspace, "cat-file", "-e", f"{approved_base}^{{commit}}")
|
|
subprocess.run(
|
|
["git", "merge-base", "--is-ancestor", approved_base, "HEAD"],
|
|
cwd=workspace,
|
|
check=True,
|
|
stdout=subprocess.DEVNULL,
|
|
stderr=subprocess.DEVNULL,
|
|
)
|
|
except (PermissionError, subprocess.CalledProcessError):
|
|
raise PermissionError("isolated_writer_approved_base_not_ancestor") from None
|
|
if _git(workspace, "merge-base", approved_base, "HEAD") != approved_base:
|
|
raise PermissionError("isolated_writer_merge_base_rejected")
|
|
common = Path(_git(workspace, "rev-parse", "--git-common-dir"))
|
|
if not common.is_absolute():
|
|
common = (workspace / common).resolve()
|
|
if common.resolve() != bare_repo.resolve():
|
|
raise PermissionError("isolated_writer_common_repo_rejected")
|
|
|
|
|
|
def approved_audit_log_path() -> Path:
|
|
root_value = os.getenv("AIOS_GITOPS_APPROVED_AUDIT_ROOT", "").strip()
|
|
log_value = os.getenv("AIOS_GITOPS_AUDIT_LOG", "").strip()
|
|
if not root_value or not log_value:
|
|
raise PermissionError("isolated_writer_audit_path_missing")
|
|
lexical_root = Path(os.path.abspath(root_value))
|
|
lexical_log = Path(os.path.abspath(log_value))
|
|
root = lexical_root.resolve(strict=True)
|
|
if lexical_root != root or root.is_symlink() or not root.is_dir():
|
|
raise PermissionError("isolated_writer_audit_root_rejected")
|
|
try:
|
|
lexical_log.relative_to(root)
|
|
except ValueError:
|
|
raise PermissionError("isolated_writer_audit_path_outside_approved_root") from None
|
|
if lexical_log.parent.resolve(strict=True) != lexical_log.parent:
|
|
raise PermissionError("isolated_writer_audit_parent_symlink_rejected")
|
|
if lexical_log.exists() and lexical_log.is_symlink():
|
|
raise PermissionError("isolated_writer_audit_log_symlink_rejected")
|
|
os.chmod(root, 0o750)
|
|
os.chmod(lexical_log.parent, 0o750)
|
|
return lexical_log
|
|
|
|
|
|
def append_audit_record(record: dict) -> None:
|
|
path = approved_audit_log_path()
|
|
flags = os.O_WRONLY | os.O_APPEND | os.O_CREAT
|
|
if hasattr(os, "O_NOFOLLOW"):
|
|
flags |= os.O_NOFOLLOW
|
|
descriptor = os.open(path, flags, 0o600)
|
|
try:
|
|
info = os.fstat(descriptor)
|
|
if not stat.S_ISREG(info.st_mode):
|
|
raise PermissionError("isolated_writer_audit_log_not_regular")
|
|
os.fchmod(descriptor, 0o600)
|
|
line = json.dumps(
|
|
_audit_sanitize(record),
|
|
ensure_ascii=False,
|
|
sort_keys=True,
|
|
separators=(",", ":"),
|
|
).encode("utf-8") + b"\n"
|
|
os.write(descriptor, line)
|
|
os.fsync(descriptor)
|
|
finally:
|
|
os.close(descriptor)
|
|
parent_fd = os.open(path.parent, os.O_RDONLY | getattr(os, "O_DIRECTORY", 0))
|
|
try:
|
|
os.fsync(parent_fd)
|
|
finally:
|
|
os.close(parent_fd)
|
|
|
|
|
|
def _audit_sanitize(value):
|
|
if isinstance(value, dict):
|
|
return {
|
|
key: (
|
|
"[REDACTED]"
|
|
if any(
|
|
marker in str(key).lower()
|
|
for marker in (
|
|
"authorization",
|
|
"credential",
|
|
"password",
|
|
"secret",
|
|
"token",
|
|
"private_key",
|
|
)
|
|
)
|
|
and str(key).lower() != "token_id"
|
|
else _audit_sanitize(inner)
|
|
)
|
|
for key, inner in value.items()
|
|
}
|
|
if isinstance(value, list):
|
|
return [_audit_sanitize(item) for item in value]
|
|
if isinstance(value, str) and re.search(
|
|
r"(?i)(authorization\s*:\s*bearer|(?:token|password|secret)\s*[:=])",
|
|
value,
|
|
):
|
|
return "[REDACTED]"
|
|
return value
|
|
|
|
|
|
def validate_endpoint(method: str, path: str, workspace: Path, bare_repo: Path) -> None:
|
|
if not enforcement_required():
|
|
return
|
|
if method != "POST" or path not in WRITE_ENDPOINTS:
|
|
raise PermissionError("isolated_writer_operation_forbidden")
|
|
require_service_invariants(workspace, bare_repo)
|
|
|
|
|
|
def validate_branch_request(data: dict) -> None:
|
|
if not enforcement_required():
|
|
return
|
|
branch = allowed_branch() if "branch" not in data else str(data.get("branch") or "").strip()
|
|
if branch != allowed_branch():
|
|
raise PermissionError("isolated_writer_branch_rejected")
|
|
if str(data.get("base", "main")).strip() != "main":
|
|
raise PermissionError("isolated_writer_base_rejected")
|
|
if data.get("force") or data.get("force_push"):
|
|
raise PermissionError("isolated_writer_force_rejected")
|
|
|
|
|
|
def validate_path(path: str, workspace: Path | None = None) -> str:
|
|
if not isinstance(path, str):
|
|
raise PermissionError("isolated_writer_path_rejected")
|
|
value = path
|
|
if not value or value != value.strip() or "\x00" in value or "\\" in value:
|
|
raise PermissionError("isolated_writer_path_rejected")
|
|
try:
|
|
decoded = unquote(value, errors="strict")
|
|
except (UnicodeDecodeError, ValueError):
|
|
raise PermissionError("isolated_writer_path_rejected") from None
|
|
if decoded != value:
|
|
raise PermissionError("isolated_writer_path_rejected")
|
|
windows = PureWindowsPath(value)
|
|
posix = PurePosixPath(value)
|
|
parts = value.split("/")
|
|
if (
|
|
posix.is_absolute()
|
|
or windows.is_absolute()
|
|
or bool(windows.drive)
|
|
or any(part in {"", ".", ".."} for part in parts)
|
|
):
|
|
raise PermissionError("isolated_writer_path_rejected")
|
|
first = parts[0]
|
|
service_exact_path = False
|
|
if first in DENIED_ROOTS:
|
|
try:
|
|
path_policy.require_service_path("commit_file.atomic", value)
|
|
service_exact_path = True
|
|
except path_policy.PathPolicyError as exc:
|
|
raise PermissionError(str(exc)) from None
|
|
if value != "README.md" and first not in ALLOWED_ROOTS and not service_exact_path:
|
|
raise PermissionError("isolated_writer_path_outside_allowed_roots")
|
|
if workspace is not None:
|
|
lexical_root = Path(os.path.abspath(str(workspace)))
|
|
root = lexical_root.resolve(strict=True)
|
|
if lexical_root != root:
|
|
raise PermissionError("isolated_writer_workspace_symlink_rejected")
|
|
current = root
|
|
for part in parts:
|
|
current = current / part
|
|
if current.is_symlink():
|
|
raise PermissionError("isolated_writer_path_symlink_rejected")
|
|
candidate = (root / value).resolve(strict=False)
|
|
try:
|
|
candidate.relative_to(root)
|
|
except ValueError:
|
|
raise PermissionError("isolated_writer_path_escapes_workspace") from None
|
|
return value
|
|
|
|
|
|
def validate_content_for_path(path: str, content: str) -> None:
|
|
if path.split("/", 1)[0] not in DENIED_ROOTS:
|
|
return
|
|
try:
|
|
path_policy.validate_content("commit_file.atomic", path, content)
|
|
except path_policy.PathPolicyError as exc:
|
|
raise PermissionError(str(exc)) from None
|
|
|
|
|
|
def configured_path_policy_sha256() -> str | None:
|
|
try:
|
|
return path_policy.configured_path_policy_sha256()
|
|
except path_policy.PathPolicyError as exc:
|
|
raise PermissionError(str(exc)) from None
|
|
|
|
|
|
def validate_commit_paths(paths: list[str], workspace: Path | None = None) -> None:
|
|
if enforcement_required():
|
|
for path in paths:
|
|
if str(path).strip() != ".":
|
|
validate_path(str(path), workspace)
|
|
|
|
|
|
def validate_staged_paths(workspace: Path) -> None:
|
|
if not enforcement_required():
|
|
return
|
|
for path in _git(workspace, "diff", "--cached", "--name-only").splitlines():
|
|
validate_path(path, workspace)
|
|
|
|
|
|
def validate_sync_branch(branch: str) -> None:
|
|
if enforcement_required() and branch != allowed_branch():
|
|
raise PermissionError("isolated_writer_sync_branch_rejected")
|