This commit is contained in:
@@ -0,0 +1,126 @@
|
||||
"""Repository discovery and filesystem containment."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
from pathlib import Path, PurePath
|
||||
import subprocess
|
||||
|
||||
from .errors import LabyricornMcpError
|
||||
|
||||
|
||||
class RepositoryGuard:
|
||||
"""Resolve every path against one verified Labyricorn Git checkout."""
|
||||
|
||||
REQUIRED_PATHS = (
|
||||
"Labyricorn.lektorproject",
|
||||
"content/articles/contents.lr",
|
||||
"content/blog/contents.lr",
|
||||
"content/projects/contents.lr",
|
||||
"content/tags/contents.lr",
|
||||
"models/entry.ini",
|
||||
"models/project.ini",
|
||||
"models/devlog-entry.ini",
|
||||
"configs/project-sources.ini",
|
||||
"scripts/build_with_projects.py",
|
||||
)
|
||||
|
||||
def __init__(self, root: Path):
|
||||
try:
|
||||
self.root = root.expanduser().resolve(strict=True)
|
||||
except OSError as exc:
|
||||
raise LabyricornMcpError(
|
||||
"REPOSITORY_NOT_FOUND", "The configured repository directory does not exist."
|
||||
) from exc
|
||||
if not self.root.is_dir():
|
||||
raise LabyricornMcpError(
|
||||
"REPOSITORY_NOT_FOUND", "The configured repository path is not a directory."
|
||||
)
|
||||
self._validate_repository()
|
||||
|
||||
@classmethod
|
||||
def discover(cls, configured: str | None = None) -> "RepositoryGuard":
|
||||
selected = configured or os.environ.get("LABYRICORN_REPOSITORY") or os.getcwd()
|
||||
return cls(Path(selected))
|
||||
|
||||
def _validate_repository(self) -> None:
|
||||
missing = [relative for relative in self.REQUIRED_PATHS if not (self.root / relative).is_file()]
|
||||
if missing:
|
||||
raise LabyricornMcpError(
|
||||
"INVALID_LABYRICORN_REPOSITORY",
|
||||
"The selected directory is not a compatible Labyricorn repository.",
|
||||
{"missing": missing},
|
||||
)
|
||||
escaped = [
|
||||
relative
|
||||
for relative in self.REQUIRED_PATHS
|
||||
if not (self.root / relative).resolve().is_relative_to(self.root)
|
||||
]
|
||||
if escaped:
|
||||
raise LabyricornMcpError(
|
||||
"INVALID_LABYRICORN_REPOSITORY",
|
||||
"Required Labyricorn paths must not resolve outside the repository.",
|
||||
{"escaped": escaped},
|
||||
)
|
||||
try:
|
||||
result = subprocess.run(
|
||||
["git", "rev-parse", "--show-toplevel"],
|
||||
cwd=self.root,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
encoding="utf-8",
|
||||
errors="replace",
|
||||
timeout=15,
|
||||
check=False,
|
||||
)
|
||||
except (OSError, subprocess.SubprocessError) as exc:
|
||||
raise LabyricornMcpError(
|
||||
"INVALID_LABYRICORN_REPOSITORY", "The selected directory is not an accessible Git repository."
|
||||
) from exc
|
||||
try:
|
||||
git_root = Path(result.stdout.strip()).resolve(strict=True)
|
||||
except OSError:
|
||||
git_root = Path()
|
||||
if result.returncode != 0 or git_root != self.root:
|
||||
raise LabyricornMcpError(
|
||||
"INVALID_LABYRICORN_REPOSITORY",
|
||||
"The selected directory must be the root of the Labyricorn Git repository.",
|
||||
)
|
||||
|
||||
def resolve(
|
||||
self,
|
||||
relative: str,
|
||||
*,
|
||||
allowed_roots: tuple[str, ...] = (),
|
||||
must_exist: bool = False,
|
||||
reject_links: bool = False,
|
||||
) -> Path:
|
||||
if not isinstance(relative, str) or not relative.strip():
|
||||
raise LabyricornMcpError("INVALID_PATH", "A non-empty repository-relative path is required.")
|
||||
raw = PurePath(relative)
|
||||
if raw.is_absolute() or raw.drive or ".." in raw.parts:
|
||||
raise LabyricornMcpError("PATH_OUTSIDE_REPOSITORY", "Only repository-relative paths are allowed.")
|
||||
candidate = self.root.joinpath(*raw.parts)
|
||||
try:
|
||||
resolved = candidate.resolve(strict=must_exist)
|
||||
except OSError as exc:
|
||||
raise LabyricornMcpError("CONTENT_NOT_FOUND", "The requested repository path does not exist.") from exc
|
||||
if not resolved.is_relative_to(self.root):
|
||||
raise LabyricornMcpError("PATH_OUTSIDE_REPOSITORY", "The requested path resolves outside the repository.")
|
||||
if allowed_roots and not any(
|
||||
resolved.is_relative_to((self.root / allowed).resolve()) for allowed in allowed_roots
|
||||
):
|
||||
raise LabyricornMcpError(
|
||||
"PATH_NOT_ALLOWED", "The requested path is outside the allowlisted repository areas.",
|
||||
{"allowed_roots": list(allowed_roots)},
|
||||
)
|
||||
if reject_links:
|
||||
current = self.root
|
||||
for part in raw.parts:
|
||||
current = current / part
|
||||
if current.exists() and (current.is_symlink() or getattr(current, "is_junction", lambda: False)()):
|
||||
raise LabyricornMcpError("PATH_LINK_REJECTED", "Linked paths cannot be modified.")
|
||||
return resolved
|
||||
|
||||
def relative(self, path: Path) -> str:
|
||||
return path.resolve().relative_to(self.root).as_posix()
|
||||
Reference in New Issue
Block a user