Files
labyricorn-site/labyricorn_mcp/safety.py
T
Labyricorn fe6025fa63
Deploy production / deploy (push) Successful in 21s
Add local repository MCP server
2026-08-23 10:48:33 -07:00

127 lines
4.9 KiB
Python

"""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()