"""Read-only, bounded access to the explicitly configured project workspace.""" from __future__ import annotations import os import re import json import difflib import hashlib import tempfile import time from pathlib import Path from pathlib import PurePosixPath from uuid import uuid4 ALLOWED_SUFFIXES = { ".py", ".dart", ".md", ".txt", ".json", ".yaml", ".yml", ".toml", ".html", ".css", ".js", ".ts", ".tsx", ".jsx", ".sh", ".ps1", } IGNORED_PARTS = { ".git", ".venv", ".dart_tool", ".video-lab", "build", "node_modules", "__pycache__", ".idea", ".vscode", } MAX_FILE_BYTES = 256 * 1024 MAX_SCAN_FILES = 500 PROPOSAL_TTL_SECONDS = 10 * 60 MAX_PENDING_PROPOSALS = 32 _pending_changes: dict[str, dict[str, object]] = {} class WorkspaceAccessDenied(ValueError): """Raised when an authenticated account selects another user's workspace.""" def configured_roots() -> tuple[Path, ...]: """Return the server administrator's allow-listed workspace roots.""" configured = os.getenv("SOVEREIGNAI_ALLOWED_WORKSPACES") if configured: values = configured.split(os.pathsep) else: primary = os.getenv("SOVEREIGNAI_WORKSPACE") values = [primary] if primary else [] roots: list[Path] = [] for value in values: if not value or not value.strip(): continue try: root = Path(value.strip()).expanduser().resolve(strict=True) except (OSError, RuntimeError): continue if root.is_dir() and root not in roots: roots.append(root) return tuple(roots) def configured_root() -> Path | None: roots = configured_roots() return roots[0] if roots else None def _roots_for_user(user_id: str | None) -> tuple[Path, ...]: roots = configured_roots() from app import auth, database registered_roots: list[Path] = [] if user_id is not None: for value in database.list_workspace_roots(user_id): try: root = Path(value).expanduser().resolve(strict=True) except (OSError, RuntimeError): continue if root.is_dir() and root not in registered_roots: registered_roots.append(root) def combine(*groups: tuple[Path, ...] | list[Path]) -> tuple[Path, ...]: combined: list[Path] = [] for group in groups: for root in group: if root not in combined: combined.append(root) return tuple(combined) if user_id is None or user_id == database.LOCAL_USER_ID: return combine(roots, registered_roots) email = auth.account_email(user_id) if email is None: raise ValueError("الحساب الحالي لا يملك مساحة عمل مخصصة على الخادم.") raw_mapping = os.getenv("SOVEREIGNAI_USER_WORKSPACES", "") try: mapping = json.loads(raw_mapping) if raw_mapping else {} except json.JSONDecodeError as exc: raise ValueError("إعداد مساحات العمل حسب الحساب غير صالح على الخادم.") from exc if not isinstance(mapping, dict): raise ValueError("إعداد مساحات العمل حسب الحساب يجب أن يكون كائن JSON.") resolved_by_email: dict[str, tuple[Path, ...]] = {} for assigned_email, assigned_values in mapping.items(): if not isinstance(assigned_email, str) or not isinstance(assigned_values, list): raise ValueError("كل مستخدم في إعداد مساحات العمل يجب أن يقابله مصفوفة مسارات.") assigned_email = assigned_email.strip().casefold() if not assigned_email or assigned_email in resolved_by_email: raise ValueError("البريد مكرر أو فارغ في إعداد مساحات العمل.") assigned_roots: list[Path] = [] for item in assigned_values: if not isinstance(item, str) or not item.strip(): raise ValueError("مسار مساحة العمل في إعداد الخادم غير صالح.") try: root = Path(item).expanduser().resolve(strict=True) except (OSError, RuntimeError) as exc: raise ValueError("أحد جذور مساحة العمل المخصصة غير موجود أو غير متاح.") from exc if not root.is_dir() or not any(root == allowed or allowed in root.parents for allowed in roots): raise ValueError("جذر الحساب يجب أن يقع داخل قائمة جذور الخادم المصرح بها.") if root in assigned_roots: raise ValueError("جذر مساحة العمل مكرر للحساب نفسه.") assigned_roots.append(root) resolved_by_email[assigned_email] = tuple(assigned_roots) # Reject overlapping account roots so one account cannot read a parent or # child directory assigned to another account. owners = list(resolved_by_email.items()) for index, (_first_email, first_roots) in enumerate(owners): for _second_email, second_roots in owners[index + 1 :]: for first_root in first_roots: for second_root in second_roots: if first_root == second_root or first_root in second_root.parents or second_root in first_root.parents: raise ValueError("جذور مساحة العمل لحسابين مختلفين متداخلة في إعداد الخادم.") assigned = resolved_by_email.get(email.casefold()) if not assigned and not registered_roots: raise WorkspaceAccessDenied("لم يخصص مسؤول الخادم مساحة عمل لهذا الحساب.") # For authenticated accounts, the global roots are validation boundaries, # not grants. Keep account access limited to its explicit assignment plus # folders that account registered from its local desktop. return combine(assigned or (), registered_roots) def validate_workspace_registration(value: str, *, must_exist: bool = True) -> Path: """Resolve a directory explicitly selected in the local desktop app.""" if not value.strip(): raise ValueError("اختر مجلد مشروع صالحًا.") try: root = Path(value).expanduser().resolve(strict=must_exist) except (OSError, RuntimeError) as exc: raise ValueError("مجلد المشروع غير موجود أو غير متاح.") from exc if must_exist and not root.is_dir(): raise ValueError("يجب اختيار مجلد صالح للمشروع.") if root == Path(root.anchor): raise ValueError("اختر مجلد مشروع محددًا، وليس جذر القرص.") if root.name.startswith(".") or root.name in IGNORED_PARTS: raise ValueError("لا يمكن تسجيل مجلد مخفي أو مستثنى كمشروع.") return root def validate_workspace_registration_isolation(root: Path, user_id: str) -> None: """Prevent locally registered roots from overlapping another tenant's admin grant.""" from app import auth, database raw_mapping = os.getenv("SOVEREIGNAI_USER_WORKSPACES", "") try: mapping = json.loads(raw_mapping) if raw_mapping else {} except json.JSONDecodeError as exc: raise ValueError("إعداد مساحات العمل حسب الحساب غير صالح على الخادم.") from exc if not isinstance(mapping, dict): raise ValueError("إعداد مساحات العمل حسب الحساب يجب أن يكون كائن JSON.") current_email = ( auth.account_email(user_id) if user_id != database.LOCAL_USER_ID else None ) candidate = os.path.normcase(os.path.abspath(str(root))) for assigned_email, assigned_paths in mapping.items(): if not isinstance(assigned_email, str) or not isinstance(assigned_paths, list): raise ValueError("كل مستخدم في إعداد مساحات العمل يجب أن يقابله مصفوفة مسارات.") if current_email and assigned_email.strip().casefold() == current_email.casefold(): continue for value in assigned_paths: if not isinstance(value, str) or not value.strip(): raise ValueError("مسار مساحة العمل المخصصة في الخادم غير صالح.") try: assigned = Path(value).expanduser().resolve(strict=True) except (OSError, RuntimeError) as exc: raise ValueError("أحد مسارات مساحة العمل المخصصة غير موجود أو غير متاح.") from exc if not assigned.is_dir(): raise ValueError("جذر مساحة العمل المخصصة ليس مجلدًا صالحًا.") normalized = os.path.normcase(os.path.abspath(str(assigned))) try: common = os.path.commonpath((candidate, normalized)) except ValueError: continue if common in (candidate, normalized): raise ValueError( "مجلد المشروع يتداخل مع مساحة عمل مخصصة لحساب آخر؛ اختر مجلدًا منفصلًا." ) def selected_root(value: str | None, *, user_id: str | None = None) -> Path | None: """Resolve a selected directory only within this user's server-assigned roots.""" allowed_roots = _roots_for_user(user_id) if value is None or not value.strip(): return allowed_roots[0] if allowed_roots else None try: root = Path(value).expanduser().resolve(strict=True) except (OSError, RuntimeError) as exc: raise ValueError("مجلد مساحة العمل المحدد غير موجود أو غير متاح.") from exc if not root.is_dir(): raise ValueError("يجب اختيار مجلد صالح لمساحة العمل.") if root == Path(root.anchor): raise ValueError("اختر مجلد مشروع محددًا، وليس جذر القرص.") if root.name.startswith(".") or root.name in IGNORED_PARTS: raise ValueError("لا يمكن استخدام مجلد مخفي أو مستثنى كمساحة عمل.") if allowed_roots and not any( root == allowed or allowed in root.parents for allowed in allowed_roots ): raise WorkspaceAccessDenied("المجلد المحدد خارج مساحة العمل المخصصة لهذا الحساب.") if not allowed_roots: raise WorkspaceAccessDenied("لا توجد جذور مساحة عمل مخصصة لهذا الحساب على الخادم.") return root def relative_file(root: Path, relative_path: str) -> Path: candidate = (root / relative_path).resolve(strict=True) try: candidate.relative_to(root) except ValueError as exc: raise ValueError("المسار المطلوب خارج مساحة العمل.") from exc if not candidate.is_file() or candidate.suffix.lower() not in ALLOWED_SUFFIXES: raise ValueError("هذا النوع من الملفات غير مسموح بقراءته.") if any( part in IGNORED_PARTS or part.startswith(".") for part in candidate.relative_to(root).parts ): raise ValueError("قراءة الملفات المخفية أو المستثناة غير مسموحة.") if candidate.stat().st_size > MAX_FILE_BYTES: raise ValueError("الملف أكبر من الحد المسموح للقراءة (256 كيلوبايت).") return candidate def relative_knowledge_file(root: Path, relative_path: str) -> Path: """Resolve a bounded UTF-8 document or PDF selected for local knowledge use.""" candidate = (root / relative_path).resolve(strict=True) try: candidate.relative_to(root) except ValueError as exc: raise ValueError("المسار المطلوب خارج مساحة العمل.") from exc if not candidate.is_file() or candidate.suffix.lower() not in (ALLOWED_SUFFIXES | {".pdf"}): raise ValueError("هذا النوع من الملفات غير مسموح بفهرسته.") if any( part in IGNORED_PARTS or part.startswith(".") for part in candidate.relative_to(root).parts ): raise ValueError("قراءة الملفات المخفية أو المستثناة غير مسموحة.") if candidate.stat().st_size > MAX_FILE_BYTES: raise ValueError("الملف أكبر من الحد المسموح للقراءة (256 كيلوبايت).") return candidate def _write_target(root: Path, relative_path: str, operation: str) -> tuple[Path, bytes]: if ( len(relative_path) > 1024 or "\x00" in relative_path or any(char in relative_path for char in '<>:"|?*') ): raise ValueError("مسار الملف غير صالح.") normalized = relative_path.replace("\\", "/") relative = PurePosixPath(normalized) if ( relative.is_absolute() or not relative.parts or any(part in {"", ".", ".."} for part in relative.parts) or any(part.startswith(".") or part in IGNORED_PARTS for part in relative.parts) ): raise ValueError("يسمح بالكتابة داخل مسارات نسبية غير مخفية في مساحة العمل فقط.") reserved_names = {"CON", "PRN", "AUX", "NUL"} | { f"{prefix}{number}" for prefix in ("COM", "LPT") for number in range(1, 10) } if any( part.endswith((".", " ")) or part.split(".", 1)[0].upper() in reserved_names for part in relative.parts ): raise ValueError("اسم الملف غير صالح على Windows.") if relative.suffix.lower() not in ALLOWED_SUFFIXES: raise ValueError("امتداد الملف غير مسموح للوكيل.") root = root.resolve(strict=True) target = root.joinpath(*relative.parts) current = root for part in relative.parts[:-1]: current = current / part if current.is_symlink(): raise ValueError("لا يسمح بالكتابة عبر مجلدات الروابط الرمزية.") if not target.parent.is_dir(): raise ValueError("يجب أن يكون المجلد الأب موجودًا؛ لا ينشئ الوكيل مجلدات تلقائيًا.") resolved_parent = target.parent.resolve(strict=True) try: resolved_parent.relative_to(root) except ValueError as exc: raise ValueError("مجلد الملف خارج مساحة العمل المحددة.") from exc if target.is_symlink(): raise ValueError("لا يسمح باستبدال ملف رابط رمزي.") exists = target.exists() if operation == "create" and exists: raise ValueError("الملف موجود بالفعل؛ اطلب تحديثه بدل إنشائه.") if operation == "update" and not exists: raise ValueError("الملف المراد تحديثه غير موجود.") if operation not in {"create", "update"}: raise ValueError("نوع التغيير غير مسموح.") if not exists: return target, b"" safe_target = relative_file(root, relative.as_posix()) raw = safe_target.read_bytes() return safe_target, raw def create_change_preview( root: Path, relative_path: str, operation: str, content: str, *, user_id: str | None = None, ) -> dict[str, object]: """Build and retain a short-lived diff; this function never writes the file.""" raw_content = content.encode("utf-8") if len(raw_content) > MAX_FILE_BYTES: raise ValueError("المحتوى المقترح يتجاوز حد 256 كيلوبايت.") target, original = _write_target(root, relative_path, operation) try: old_text = original.decode("utf-8") except UnicodeDecodeError as exc: raise ValueError("لا يمكن تحديث ملف غير محفوظ بترميز UTF-8.") from exc newline = "\r\n" if b"\r\n" in original else "\n" proposed_text = content.replace("\r\n", "\n").replace("\n", newline) proposed_bytes = proposed_text.encode("utf-8") if operation == "update" and original == proposed_bytes: raise ValueError("المحتوى المقترح مطابق للملف الحالي ولا يحتاج إلى تعديل.") relative = relative_path.replace("\\", "/") before = old_text.splitlines(keepends=True) after = proposed_text.splitlines(keepends=True) diff = "".join( difflib.unified_diff( before, after, fromfile=f"a/{relative}" if operation == "update" else "/dev/null", tofile=f"b/{relative}", lineterm="", ) ) expired = [ key for key, item in _pending_changes.items() if float(item["expires_at"]) <= time.time() ] for key in expired: _pending_changes.pop(key, None) if len(_pending_changes) >= MAX_PENDING_PROPOSALS: raise ValueError("هناك عدد كبير من معاينات التغيير المعلقة؛ ألغِ بعضها قبل إنشاء معاينة أخرى.") token = str(uuid4()) _pending_changes[token] = { "root": str(root.resolve(strict=True)), "path": relative, "operation": operation, "content": proposed_bytes, "expected_hash": hashlib.sha256(original).hexdigest(), "expires_at": time.time() + PROPOSAL_TTL_SECONDS, "user_id": user_id, } return { "token": token, "path": relative, "operation": operation, "diff": diff, "expires_in_seconds": PROPOSAL_TTL_SECONDS, } def apply_change_preview( token: str, *, user_id: str | None = None ) -> dict[str, str]: """Apply a reviewed proposal once, only if its target is still unchanged.""" proposal = _pending_changes.get(token) if proposal is None or float(proposal["expires_at"]) <= time.time(): _pending_changes.pop(token, None) raise ValueError("انتهت صلاحية معاينة التغيير أو استُخدمت مسبقًا؛ أنشئ معاينة جديدة.") if proposal.get("user_id") != user_id: raise ValueError("معاينة التعديل لا تخص جلسة المستخدم الحالية.") _pending_changes.pop(token, None) root = Path(str(proposal["root"])).resolve(strict=True) relative_path = str(proposal["path"]) operation = str(proposal["operation"]) target, current = _write_target(root, relative_path, operation) current_hash = hashlib.sha256(current).hexdigest() if current_hash != proposal["expected_hash"]: raise ValueError("تغير الملف منذ عرض المعاينة؛ أنشئ diff جديدًا قبل التطبيق.") content = proposal["content"] if not isinstance(content, bytes): raise ValueError("بيانات المعاينة غير صالحة.") temporary_path: str | None = None try: with tempfile.NamedTemporaryFile( mode="wb", prefix=".sovereignai-review-", suffix=".tmp", dir=target.parent, delete=False, ) as temporary_file: temporary_path = temporary_file.name temporary_file.write(content) temporary_file.flush() os.fsync(temporary_file.fileno()) if operation == "update": os.chmod(temporary_path, target.stat().st_mode) # Recheck to avoid clobbering edits made while the temporary file was written. _, latest = _write_target(root, relative_path, operation) if hashlib.sha256(latest).hexdigest() != proposal["expected_hash"]: raise ValueError("تغير الملف أثناء التطبيق؛ لم تُحفظ المعاينة.") os.replace(temporary_path, target) temporary_path = None finally: if temporary_path is not None: try: os.unlink(temporary_path) except OSError: pass return {"path": relative_path, "operation": operation, "status": "applied"} def list_text_files(root: Path) -> list[Path]: files: list[Path] = [] for current, directories, filenames in os.walk(root, followlinks=False): directories[:] = [ name for name in directories if name not in IGNORED_PARTS and not name.startswith(".") ] for filename in filenames: path = Path(current) / filename if path.suffix.lower() not in ALLOWED_SUFFIXES: continue try: relative_file(root, path.relative_to(root).as_posix()) except (OSError, ValueError): continue files.append(path) if len(files) >= MAX_SCAN_FILES: return files return files def list_knowledge_files(root: Path) -> list[Path]: files: list[Path] = [] for current, directories, filenames in os.walk(root, followlinks=False): directories[:] = [ name for name in directories if name not in IGNORED_PARTS and not name.startswith(".") ] for filename in filenames: path = Path(current) / filename if path.suffix.lower() not in (ALLOWED_SUFFIXES | {".pdf"}): continue try: relative_knowledge_file(root, path.relative_to(root).as_posix()) except (OSError, ValueError): continue files.append(path) if len(files) >= MAX_SCAN_FILES: return files return files def retrieve(task: str, root: Path, limit: int = 3) -> list[tuple[str, str]]: terms = { term.casefold() for term in re.findall(r"[\w\u0600-\u06ff]{3,}", task) if term.casefold() not in { "the", "and", "for", "with", "this", "that", "من", "على", "في", "عن", "كيف", "شو", "ما", "ماذا", "هذا", "هذه", "التي", "الذي", "اشرح", "دور", "ملف", "ملفات", "اذكر", "المستخدمة", "المستخدم", "المشروع", "التفسير", } } mentioned_paths = { match.replace("\\", "/").casefold() for match in re.findall( r"(?:[\w.-]+[\\/])+[\w.-]+\.(?:py|dart|md|txt|json|ya?ml|toml|html|css|js|ts|tsx|jsx|sh|ps1)", task, flags=re.IGNORECASE, ) } ranked: list[tuple[int, str, str]] = [] for path in list_text_files(root): try: text = path.read_text(encoding="utf-8", errors="replace") except OSError: continue if not text.strip(): continue lowered = text.casefold() words = re.findall(r"[\w\u0600-\u06ff]+", lowered) score = sum(words.count(term) for term in terms) relative = path.relative_to(root).as_posix() if relative.casefold() in mentioned_paths: score += 100_000 if score: # If the user named a source file explicitly, include a larger bounded # excerpt so the agent can explain the code rather than just locate it. excerpt_limit = 12_000 if relative.casefold() in mentioned_paths else 1_800 ranked.append((score, relative, text[:excerpt_limit])) ranked.sort(key=lambda item: (-item[0], item[1])) explicit_matches = [ item for item in ranked if item[1].casefold() in mentioned_paths ] if explicit_matches: ranked = explicit_matches return [(relative, content) for _, relative, content in ranked[:limit]]