219 lines
8.7 KiB
Python
219 lines
8.7 KiB
Python
"""Build a bounded, secret-aware project snapshot for isolated command runs."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import os
|
|
import re
|
|
import stat
|
|
import tempfile
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
from pathlib import PurePosixPath
|
|
from typing import Any
|
|
|
|
from app import workspace
|
|
|
|
MAX_SNAPSHOT_FILES = 50
|
|
MAX_SNAPSHOT_FILE_BYTES = 512 * 1024
|
|
MAX_SNAPSHOT_TOTAL_BYTES = 10 * 1024 * 1024
|
|
_SENSITIVE_NAME = re.compile(
|
|
r"(^|[._-])(secrets?|credentials?|passwords?|tokens?|api[_-]?keys?)([._-]|$)",
|
|
re.IGNORECASE,
|
|
)
|
|
_WINDOWS_RESERVED_NAMES = {"CON", "PRN", "AUX", "NUL"} | {
|
|
f"{prefix}{number}"
|
|
for prefix in ("COM", "LPT")
|
|
for number in range(1, 10)
|
|
}
|
|
_KNOWN_SECRET = re.compile(
|
|
r"gsk_[A-Za-z0-9]{20,}|sk-[A-Za-z0-9_-]{20,}|"
|
|
r"gh[pousr]_[A-Za-z0-9]{20,}|github_pat_[A-Za-z0-9_]{20,}|"
|
|
r"AKIA[0-9A-Z]{16}|-----BEGIN (?:RSA |EC |OPENSSH )?PRIVATE KEY-----"
|
|
)
|
|
_SECRET_ASSIGNMENT = re.compile(
|
|
r"(?im)\b(?:api[_-]?key|secret(?:[_-]?key)?|password|"
|
|
r"access[_-]?token|auth[_-]?token|private[_-]?key)\b\s*[:=]\s*"
|
|
r"[\"'](?P<value>[^\"'\r\n]{12,})[\"']"
|
|
)
|
|
_PLACEHOLDER_VALUES = {
|
|
"your_api_key_here",
|
|
"your-secret-here",
|
|
"replace-me",
|
|
"changeme",
|
|
"placeholder",
|
|
}
|
|
|
|
|
|
@dataclass
|
|
class StagedWorkspace:
|
|
"""Temporary copy of explicit, bounded workspace files; call close() when done."""
|
|
|
|
root: Path
|
|
files: list[dict[str, Any]]
|
|
total_bytes: int
|
|
_temporary: tempfile.TemporaryDirectory
|
|
|
|
def close(self) -> None:
|
|
self._temporary.cleanup()
|
|
|
|
def __enter__(self) -> "StagedWorkspace":
|
|
return self
|
|
|
|
def __exit__(self, *_exc: object) -> None:
|
|
self.close()
|
|
|
|
|
|
def _is_reparse_point(path: Path) -> bool:
|
|
if path.is_symlink():
|
|
return True
|
|
is_junction = getattr(path, "is_junction", None)
|
|
return bool(is_junction()) if is_junction is not None else False
|
|
|
|
|
|
def _contains_credential(content: bytes) -> bool:
|
|
text = content.decode("utf-8", errors="replace")
|
|
if _KNOWN_SECRET.search(text):
|
|
return True
|
|
for match in _SECRET_ASSIGNMENT.finditer(text):
|
|
value = match.group("value").strip()
|
|
if value.casefold() in _PLACEHOLDER_VALUES or value.startswith(("$", "${", "<")):
|
|
continue
|
|
return True
|
|
return False
|
|
|
|
|
|
def _safe_relative_path(root: Path, value: str) -> tuple[Path, str]:
|
|
if (
|
|
not isinstance(value, str)
|
|
or not value
|
|
or len(value) > 240
|
|
or "\x00" in value
|
|
or any(ord(char) < 32 for char in value)
|
|
):
|
|
raise ValueError("مسار الملف المحدد غير صالح.")
|
|
normalized = value.replace("\\", "/")
|
|
parts = normalized.split("/")
|
|
relative = PurePosixPath(normalized)
|
|
if (
|
|
relative.is_absolute()
|
|
or not parts
|
|
or any(part in {"", ".", ".."} for part in parts)
|
|
or any(":" in part or any(char in part for char in '<>|?*"') for part in parts)
|
|
or any(
|
|
len(part) > 255
|
|
or part.endswith((".", " "))
|
|
or part.split(".", 1)[0].upper() in _WINDOWS_RESERVED_NAMES
|
|
for part in parts
|
|
)
|
|
or any(part.startswith(".") or part in workspace.IGNORED_PARTS for part in parts)
|
|
):
|
|
raise ValueError("لا يسمح بنسخ مسارات مطلقة أو مخفية أو مستثناة خارج الملف المحدد.")
|
|
if not relative.parts or relative.suffix.lower() not in workspace.ALLOWED_SUFFIXES:
|
|
raise ValueError("امتداد الملف غير مسموح لنسخة التنفيذ.")
|
|
if any(_SENSITIVE_NAME.search(part) for part in parts):
|
|
raise ValueError("اسم الملف يوحي باحتوائه على بيانات سرية؛ لم تتم إضافته إلى نسخة التنفيذ.")
|
|
|
|
candidate = root.joinpath(*relative.parts)
|
|
current = root
|
|
try:
|
|
for part in relative.parts:
|
|
current = current / part
|
|
if _is_reparse_point(current):
|
|
raise ValueError("نسخ الملفات عبر الروابط الرمزية أو junctions غير مسموح.")
|
|
resolved = candidate.resolve(strict=True)
|
|
resolved.relative_to(root)
|
|
metadata = os.stat(resolved, follow_symlinks=False)
|
|
except ValueError:
|
|
raise
|
|
except (OSError, RuntimeError) as exc:
|
|
raise ValueError("الملف المحدد غير موجود أو غير متاح داخل مساحة العمل.") from exc
|
|
if not stat.S_ISREG(metadata.st_mode):
|
|
raise ValueError("يمكن نسخ الملفات العادية فقط.")
|
|
if metadata.st_size > MAX_SNAPSHOT_FILE_BYTES:
|
|
raise ValueError("الملف يتجاوز حد 512 كيلوبايت لنسخة التنفيذ.")
|
|
return resolved, relative.as_posix()
|
|
|
|
|
|
def stage_selected_files(
|
|
root: Path, relative_paths: list[str], *, user_id: str | None = None
|
|
) -> StagedWorkspace:
|
|
"""Copy only selected safe files into a fresh temporary staging directory."""
|
|
resolved_root = workspace.selected_root(str(root), user_id=user_id)
|
|
if resolved_root is None:
|
|
raise ValueError("لم يتم اختيار مساحة عمل مصرح بها.")
|
|
if not resolved_root.is_dir():
|
|
raise ValueError("مساحة العمل المحددة ليست مجلدًا.")
|
|
if not isinstance(relative_paths, list) or not relative_paths:
|
|
raise ValueError("حدد ملفًا واحدًا على الأقل لنسخة التنفيذ.")
|
|
if len(relative_paths) > MAX_SNAPSHOT_FILES:
|
|
raise ValueError("نسخة التنفيذ تقبل حتى 50 ملفًا في الطلب الواحد.")
|
|
|
|
selected: list[tuple[Path, str, os.stat_result]] = []
|
|
seen: set[str] = set()
|
|
total_bytes = 0
|
|
for raw_path in relative_paths:
|
|
source, relative = _safe_relative_path(resolved_root, raw_path)
|
|
key = relative.casefold()
|
|
if key in seen:
|
|
raise ValueError("قائمة ملفات نسخة التنفيذ تحتوي مسارًا مكررًا.")
|
|
seen.add(key)
|
|
metadata = source.stat()
|
|
total_bytes += metadata.st_size
|
|
if total_bytes > MAX_SNAPSHOT_TOTAL_BYTES:
|
|
raise ValueError("إجمالي نسخة التنفيذ يتجاوز 10 ميغابايت.")
|
|
selected.append((source, relative, metadata))
|
|
|
|
staged_records: list[dict[str, Any]] = []
|
|
temporary = tempfile.TemporaryDirectory(prefix="sovereignai-execution-")
|
|
stage_root = Path(temporary.name).resolve(strict=True)
|
|
try:
|
|
copied_bytes = 0
|
|
for source, relative, expected in selected:
|
|
with source.open("rb") as input_file:
|
|
opened = os.fstat(input_file.fileno())
|
|
if (opened.st_dev, opened.st_ino, opened.st_size) != (
|
|
expected.st_dev,
|
|
expected.st_ino,
|
|
expected.st_size,
|
|
):
|
|
raise ValueError("تغير الملف أثناء تجهيز نسخة التنفيذ؛ أعد المحاولة.")
|
|
content = input_file.read(MAX_SNAPSHOT_FILE_BYTES + 1)
|
|
if len(content) > MAX_SNAPSHOT_FILE_BYTES or len(content) != expected.st_size:
|
|
raise ValueError("تغير حجم الملف أثناء تجهيز نسخة التنفيذ؛ أعد المحاولة.")
|
|
if _contains_credential(content):
|
|
raise ValueError(
|
|
f"الملف {relative} يبدو أنه يحتوي على مفتاح أو قيمة اعتماد؛ لم تتم إضافته."
|
|
)
|
|
latest = source.stat()
|
|
if (latest.st_dev, latest.st_ino, latest.st_size) != (
|
|
expected.st_dev,
|
|
expected.st_ino,
|
|
expected.st_size,
|
|
):
|
|
raise ValueError("تغير الملف أثناء تجهيز نسخة التنفيذ؛ أعد المحاولة.")
|
|
|
|
destination = stage_root.joinpath(*PurePosixPath(relative).parts)
|
|
destination.parent.mkdir(parents=True, exist_ok=True)
|
|
with destination.open("xb") as output_file:
|
|
output_file.write(content)
|
|
copied_bytes += len(content)
|
|
staged_records.append(
|
|
{
|
|
"path": relative,
|
|
"size_bytes": len(content),
|
|
"sha256": hashlib.sha256(content).hexdigest(),
|
|
}
|
|
)
|
|
if copied_bytes != total_bytes:
|
|
raise ValueError("تغير حجم نسخة الملفات أثناء تجهيزها؛ أعد المحاولة.")
|
|
return StagedWorkspace(
|
|
root=stage_root,
|
|
files=staged_records,
|
|
total_bytes=copied_bytes,
|
|
_temporary=temporary,
|
|
)
|
|
except Exception:
|
|
temporary.cleanup()
|
|
raise
|