1261 lines
44 KiB
Python
1261 lines
44 KiB
Python
"""Read-only Git state surface for the Hermes-Relay plugin.
|
|
|
|
Discovers repositories under a configurable base path (default ``~/projects``)
|
|
and exposes working-tree status, branches, per-file diffs, and tracked-file
|
|
reads. All git execution uses argument lists with ``git -C <repo>`` and never
|
|
shell interpolation. Every operation is bounded (timeouts + size caps) and every
|
|
``repo`` reference is validated against the scanned allowlist.
|
|
|
|
Security contract
|
|
-----------------
|
|
- ``repo`` params are opaque ids resolved against the scanned repo set; an
|
|
unknown id is rejected before any filesystem access.
|
|
- File paths are validated to reject traversal (``..``), absolute escapes, and
|
|
null bytes. Working-tree reads additionally require canonical containment.
|
|
- Remote URLs are scrubbed of embedded userinfo before they reach any client.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import os
|
|
import re
|
|
import stat
|
|
import subprocess
|
|
import tempfile
|
|
from functools import wraps
|
|
from pathlib import Path
|
|
from threading import Lock, RLock
|
|
from typing import Any
|
|
|
|
from .config import raw_config_value
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# Bounded response caps (bytes / entries). Configurable via env for operators.
|
|
MAX_STATUS_ENTRIES = int(os.environ.get("GIT_STATE_MAX_STATUS_ENTRIES", "200"))
|
|
MAX_DIFF_BYTES = int(os.environ.get("GIT_STATE_MAX_DIFF_BYTES", "64_000"))
|
|
MAX_FILE_BYTES = int(os.environ.get("GIT_STATE_MAX_FILE_BYTES", "256_000"))
|
|
MAX_GIT_OUTPUT_BYTES = int(os.environ.get("GIT_STATE_MAX_OUTPUT_BYTES", "1_000_000"))
|
|
MAX_GIT_ERROR_BYTES = int(os.environ.get("GIT_STATE_MAX_ERROR_BYTES", "16_000"))
|
|
MAX_GIT_SCALAR_LENGTH = 512
|
|
GIT_TIMEOUT_SECONDS = float(os.environ.get("GIT_STATE_TIMEOUT_SECONDS", "10"))
|
|
|
|
# Default base path for repo discovery.
|
|
DEFAULT_BASE_PATH = "~/projects"
|
|
_BASE_PATH_ENV = "GIT_STATE_BASE_PATH"
|
|
|
|
# Matches a remote URL's userinfo (user[:password]@) so it can be scrubbed.
|
|
_USERINFO_RE = re.compile(r"^([a-zA-Z][a-zA-Z0-9+.-]*://)([^/@]+)@")
|
|
_SSH_USERINFO_RE = re.compile(r"^([^/@:]+)@([^:]+):")
|
|
_ERROR_USERINFO_RE = re.compile(r"([a-zA-Z][a-zA-Z0-9+.-]*://)([^\s/@]+)@")
|
|
_ERROR_SSH_USERINFO_RE = re.compile(r"(?<![\w@])([^\s/@:]+)@([^\s:]+):")
|
|
|
|
# FastAPI executes synchronous routes concurrently. These locks serialize Git
|
|
# operations per repository without creating worker threads, so one API call
|
|
# cannot replace a validated working-tree path underneath another.
|
|
_REPO_LOCKS_GUARD = Lock()
|
|
_REPO_LOCKS: dict[str, RLock] = {}
|
|
|
|
|
|
class GitStateError(ValueError):
|
|
"""A caller supplied an invalid repo id, path, or diff kind."""
|
|
|
|
|
|
class GitError(GitStateError):
|
|
"""A mutation failed in a way git or the security gate reported.
|
|
|
|
``code`` is a stable, machine-readable taxonomy tag the UI can map to a
|
|
readable message without ever rendering a raw stack trace or JSON dump:
|
|
``non-repo``, ``dirty``, ``conflict``, ``auth``, ``network``,
|
|
``invalid-input``, ``missing-confirmation``, ``wrong-confirmation``.
|
|
"""
|
|
|
|
_CODES = {
|
|
"non-repo",
|
|
"dirty",
|
|
"conflict",
|
|
"auth",
|
|
"network",
|
|
"invalid-input",
|
|
"missing-confirmation",
|
|
"wrong-confirmation",
|
|
}
|
|
|
|
def __init__(self, message: str, code: str = "invalid-input") -> None:
|
|
if code not in self._CODES:
|
|
raise ValueError(f"unknown git error code: {code}")
|
|
super().__init__(message)
|
|
self.code = code
|
|
|
|
|
|
def base_path() -> Path:
|
|
"""Resolve the configured discovery base path (default ``~/projects``)."""
|
|
raw = raw_config_value(_BASE_PATH_ENV) or DEFAULT_BASE_PATH
|
|
return Path(raw).expanduser()
|
|
|
|
|
|
def _run_git_bounded(
|
|
repo: Path,
|
|
args: list[str],
|
|
*,
|
|
mutation: bool,
|
|
literal_pathspecs: bool = False,
|
|
) -> tuple[int, str, str]:
|
|
"""Run Git without materializing unbounded stdout or stderr in memory."""
|
|
error_type = GitError if mutation else GitStateError
|
|
with tempfile.TemporaryFile() as stdout_file, tempfile.TemporaryFile() as stderr_file:
|
|
try:
|
|
command = ["git", "-C", str(repo)]
|
|
if literal_pathspecs:
|
|
command.append("--literal-pathspecs")
|
|
command.extend(args)
|
|
result = subprocess.run(
|
|
command,
|
|
stdout=stdout_file,
|
|
stderr=stderr_file,
|
|
timeout=GIT_TIMEOUT_SECONDS,
|
|
check=False,
|
|
)
|
|
except subprocess.TimeoutExpired as exc:
|
|
if mutation:
|
|
raise GitError(f"git timed out for {repo.name}", code="network") from exc
|
|
raise GitStateError(f"git timed out for {repo.name}") from exc
|
|
except OSError as exc:
|
|
if mutation:
|
|
raise GitError(
|
|
f"could not run git for {repo.name}: {exc}",
|
|
code="non-repo",
|
|
) from exc
|
|
raise GitStateError(f"could not run git for {repo.name}: {exc}") from exc
|
|
|
|
stdout_file.seek(0)
|
|
stderr_file.seek(0)
|
|
stdout_bytes = stdout_file.read(MAX_GIT_OUTPUT_BYTES + 1)
|
|
stderr_bytes = stderr_file.read(MAX_GIT_ERROR_BYTES + 1)
|
|
if len(stdout_bytes) > MAX_GIT_OUTPUT_BYTES:
|
|
message = f"git {args[0] if args else 'command'} output exceeded the limit"
|
|
if mutation:
|
|
raise GitError(message, code="invalid-input")
|
|
raise error_type(message)
|
|
|
|
stdout = stdout_bytes.decode("utf-8", errors="replace")
|
|
stderr = stderr_bytes[:MAX_GIT_ERROR_BYTES].decode("utf-8", errors="replace")
|
|
if len(stderr_bytes) > MAX_GIT_ERROR_BYTES:
|
|
stderr += "\n[error output truncated]"
|
|
return result.returncode, stdout, stderr
|
|
|
|
|
|
def _safe_git_error(text: str) -> str:
|
|
"""Bound and scrub URL userinfo before returning Git diagnostics."""
|
|
scrubbed = _ERROR_USERINFO_RE.sub(r"\1", text)
|
|
scrubbed = _ERROR_SSH_USERINFO_RE.sub(r"\2:", scrubbed)
|
|
return scrubbed[:MAX_GIT_ERROR_BYTES].strip()
|
|
|
|
|
|
def _git(repo: Path, *args: str, literal_pathspecs: bool = False) -> str:
|
|
"""Run ``git -C <repo> <args>`` and return bounded stdout."""
|
|
returncode, stdout, stderr = _run_git_bounded(
|
|
repo,
|
|
list(args),
|
|
mutation=False,
|
|
literal_pathspecs=literal_pathspecs,
|
|
)
|
|
if returncode != 0:
|
|
raise GitStateError(
|
|
f"git {args[0] if args else 'command'} failed for {repo.name}: "
|
|
f"{_safe_git_error(stderr or stdout)}"
|
|
)
|
|
return stdout
|
|
|
|
|
|
def _is_git_repo(path: Path) -> bool:
|
|
"""True if ``path`` is a git work tree (has a .git entry)."""
|
|
return (path / ".git").exists()
|
|
|
|
|
|
def _is_link_or_junction(path: Path) -> bool:
|
|
is_junction = getattr(path, "is_junction", None)
|
|
if path.is_symlink() or bool(is_junction and is_junction()):
|
|
return True
|
|
if os.name != "nt":
|
|
return False
|
|
try:
|
|
attributes = os.lstat(path).st_file_attributes
|
|
except (AttributeError, FileNotFoundError, OSError):
|
|
return False
|
|
reparse_point = getattr(stat, "FILE_ATTRIBUTE_REPARSE_POINT", 0x400)
|
|
return bool(attributes & reparse_point)
|
|
|
|
|
|
def _is_within(root: Path, candidate: Path) -> bool:
|
|
return candidate == root or root in candidate.parents
|
|
|
|
|
|
def _canonical_repo_root(repo: Path) -> Path:
|
|
"""Return an unlinked absolute repository root or fail closed."""
|
|
lexical = Path(os.path.abspath(repo))
|
|
try:
|
|
if _is_link_or_junction(lexical):
|
|
raise GitStateError("repository root changed during operation")
|
|
canonical = lexical.resolve(strict=True)
|
|
except GitStateError:
|
|
raise
|
|
except (OSError, RuntimeError) as exc:
|
|
raise GitStateError("repository root changed during operation") from exc
|
|
if os.path.normcase(str(canonical)) != os.path.normcase(str(lexical)):
|
|
raise GitStateError("repository root changed during operation")
|
|
return canonical
|
|
|
|
|
|
def _serialized_repo_operation(function: Any) -> Any:
|
|
"""Serialize a Git operation against other operations on the same repo."""
|
|
|
|
@wraps(function)
|
|
def wrapped(repo: Path, *args: Any, **kwargs: Any) -> Any:
|
|
key = os.path.normcase(os.path.abspath(repo))
|
|
with _REPO_LOCKS_GUARD:
|
|
lock = _REPO_LOCKS.setdefault(key, RLock())
|
|
with lock:
|
|
return function(repo, *args, **kwargs)
|
|
|
|
return wrapped
|
|
|
|
|
|
def _resolve_repo_disk_path(
|
|
repo: Path,
|
|
path: str,
|
|
*,
|
|
strict: bool,
|
|
) -> Path:
|
|
"""Resolve a validated relative path and prove canonical repo containment.
|
|
|
|
``Path.resolve`` follows symlinks and junctions before ``commonpath``
|
|
compares the canonical filesystem paths. Different-drive paths fail
|
|
closed on Windows. Callers that open the result must revalidate the opened
|
|
handle before using it as defense in depth against an unexpected path
|
|
replacement. Concurrent external same-user filesystem mutation is outside
|
|
the plugin trust boundary.
|
|
"""
|
|
safe_path = resolve_repo_path(repo, path)
|
|
root = _canonical_repo_root(repo)
|
|
candidate = (root / safe_path).resolve(strict=strict)
|
|
try:
|
|
common = Path(os.path.commonpath((str(root), str(candidate))))
|
|
except ValueError as exc:
|
|
raise GitStateError(f"path escapes repository: {safe_path}") from exc
|
|
if os.path.normcase(str(common)) != os.path.normcase(str(root)):
|
|
raise GitStateError(f"path escapes repository: {safe_path}")
|
|
return candidate
|
|
|
|
|
|
def _has_link_component(base: Path, path: Path) -> bool:
|
|
current = base
|
|
try:
|
|
relative = path.relative_to(base)
|
|
except ValueError:
|
|
return True
|
|
for part in relative.parts:
|
|
current /= part
|
|
if _is_link_or_junction(current):
|
|
return True
|
|
return False
|
|
|
|
|
|
def _validate_untracked_delete_path(repo: Path, path: str) -> str:
|
|
"""Validate one exact untracked file before delegating deletion to Git."""
|
|
safe_path = resolve_repo_path(repo, path)
|
|
root = _canonical_repo_root(repo)
|
|
lexical = root / safe_path
|
|
if _has_link_component(root, lexical):
|
|
raise GitStateError(f"path contains a link or junction: {safe_path}")
|
|
candidate = _resolve_repo_disk_path(repo, safe_path, strict=False)
|
|
if candidate.exists() and candidate.is_dir():
|
|
raise GitStateError(f"path is not a file: {safe_path}")
|
|
return safe_path
|
|
|
|
|
|
def repo_id(repo: Path, base: Path | None = None) -> str:
|
|
"""Stable collision-free id relative to the configured canonical base."""
|
|
if base is None:
|
|
return repo.name
|
|
return repo.relative_to(base).as_posix()
|
|
|
|
|
|
def scan_repos(base: Path) -> list[dict[str, Any]]:
|
|
"""Recursively scan ``base`` for git repositories.
|
|
|
|
Returns a list of ``{id, name, root, current_branch, dirty}``. A missing
|
|
base path yields an empty list (never an exception). ``.git`` internals are
|
|
never reported as repositories.
|
|
"""
|
|
if not base.is_dir():
|
|
return []
|
|
|
|
canonical_base = base.resolve()
|
|
repos: list[dict[str, Any]] = []
|
|
for root in sorted(base.rglob("*")):
|
|
if not root.is_dir():
|
|
continue
|
|
if root.name == ".git":
|
|
continue
|
|
if _has_link_component(base, root):
|
|
continue
|
|
canonical_root = root.resolve()
|
|
if not _is_within(canonical_base, canonical_root):
|
|
continue
|
|
if not _is_git_repo(canonical_root):
|
|
continue
|
|
repos.append(_describe_repo(canonical_root, canonical_base))
|
|
return repos
|
|
|
|
|
|
def _describe_repo(repo: Path, base: Path) -> dict[str, Any]:
|
|
"""Build the scan entry for one repository."""
|
|
current_branch = ""
|
|
try:
|
|
current_branch = _git(repo, "symbolic-ref", "--short", "-q", "HEAD").strip()
|
|
except GitStateError:
|
|
# Detached HEAD or unborn branch — leave empty.
|
|
pass
|
|
|
|
dirty = False
|
|
try:
|
|
dirty = bool(_git(repo, "status", "--porcelain").strip())
|
|
except GitStateError:
|
|
pass
|
|
|
|
return {
|
|
"id": repo_id(repo, base),
|
|
"name": repo.name,
|
|
"root": str(repo),
|
|
"current_branch": current_branch,
|
|
"dirty": dirty,
|
|
}
|
|
|
|
|
|
def _scan_index(base: Path) -> dict[str, Path]:
|
|
"""Build the id → root allowlist from a scan."""
|
|
return {entry["id"]: Path(entry["root"]) for entry in scan_repos(base)}
|
|
|
|
|
|
def resolve_repo(base: Path, repo: str) -> Path:
|
|
"""Resolve a ``repo`` id against the scanned allowlist.
|
|
|
|
Raises [GitStateError] if the id is unknown (never accepts an arbitrary
|
|
path).
|
|
"""
|
|
index = _scan_index(base)
|
|
root = index.get(repo)
|
|
if root is None:
|
|
raise GitStateError(f"unknown repository: {repo}")
|
|
return root
|
|
|
|
|
|
def resolve_repo_path(repo: Path, path: str) -> str:
|
|
"""Validate a repo-relative file path, rejecting traversal and escapes.
|
|
|
|
Returns the normalized path. Raises [GitStateError] on any unsafe input.
|
|
"""
|
|
if not isinstance(path, str) or not path:
|
|
raise GitStateError("path is required")
|
|
if "\x00" in path:
|
|
raise GitStateError("path contains a null byte")
|
|
if "%" in path:
|
|
# Reject URL-encoding artifacts; the framework may decode %2F to a
|
|
# path separator, so a literal % is never safe in a repo path.
|
|
raise GitStateError("path contains a percent-encoding artifact")
|
|
normalized = path.replace("\\", "/")
|
|
if normalized.startswith("/"):
|
|
raise GitStateError("absolute paths are not allowed")
|
|
if normalized.startswith("~"):
|
|
raise GitStateError("home-relative paths are not allowed")
|
|
segments = normalized.split("/")
|
|
if any(seg in ("", ".", "..") for seg in segments):
|
|
raise GitStateError("path traversal is not allowed")
|
|
return normalized
|
|
|
|
|
|
def repo_status(repo: Path) -> dict[str, Any]:
|
|
"""Return grouped working-tree status with counts and truncation flag."""
|
|
porcelain = _git(repo, "status", "--porcelain=v1", "-z")
|
|
staged: list[dict[str, Any]] = []
|
|
modified: list[dict[str, Any]] = []
|
|
untracked: list[dict[str, Any]] = []
|
|
truncated = False
|
|
|
|
def _numstat(*args: str) -> dict[str, tuple[int, int]]:
|
|
"""Return bounded text-line deltas keyed by the destination path.
|
|
|
|
``--numstat -z`` keeps tabs/newlines in filenames unambiguous. Binary
|
|
files report ``-`` for each count and intentionally remain without a
|
|
line delta in the UI.
|
|
"""
|
|
raw = _git(repo, "diff", "--numstat", "-z", *args)
|
|
values = raw.split("\0")
|
|
result: dict[str, tuple[int, int]] = {}
|
|
index = 0
|
|
while index < len(values):
|
|
record = values[index]
|
|
index += 1
|
|
if not record:
|
|
continue
|
|
fields = record.split("\t", 2)
|
|
if len(fields) != 3:
|
|
continue
|
|
added, deleted, path = fields
|
|
if not path:
|
|
# Rename/copy records place old and new paths in the next two
|
|
# NUL-delimited fields. Status exposes the destination path.
|
|
index += 1
|
|
if index >= len(values):
|
|
break
|
|
path = values[index]
|
|
index += 1
|
|
if added.isdigit() and deleted.isdigit():
|
|
result[path] = (int(added), int(deleted))
|
|
return result
|
|
|
|
staged_numstat = _numstat("--cached", "--")
|
|
modified_numstat = _numstat("--")
|
|
|
|
# -z separates records with NUL; each record is "<XY> <path>\0". A rename
|
|
# or copy emits TWO records ("R new\0old\0"); the bare source-path record
|
|
# must be skipped, not misparsed as an XY record.
|
|
records = porcelain.split("\0")
|
|
i = 0
|
|
while i < len(records):
|
|
record = records[i]
|
|
i += 1
|
|
if not record:
|
|
continue
|
|
xy = record[:2]
|
|
path = record[3:]
|
|
if not path:
|
|
continue
|
|
if xy == "??":
|
|
untracked.append({"path": path, "additions": None, "deletions": None})
|
|
# Independent checks: a file staged AND modified lands in both groups.
|
|
if xy[0] in ("M", "A", "D", "R", "C"):
|
|
additions, deletions = staged_numstat.get(path, (None, None))
|
|
staged.append({"path": path, "additions": additions, "deletions": deletions})
|
|
if xy[1] in ("M", "D"):
|
|
additions, deletions = modified_numstat.get(path, (None, None))
|
|
modified.append({"path": path, "additions": additions, "deletions": deletions})
|
|
if xy[0] in ("R", "C"):
|
|
# The immediately following record is the rename/copy source path;
|
|
# skip it so it is not misparsed as an XY record.
|
|
i += 1
|
|
|
|
def _bounded(items: list[dict[str, Any]]) -> list[dict[str, Any]]:
|
|
nonlocal truncated
|
|
if len(items) > MAX_STATUS_ENTRIES:
|
|
truncated = True
|
|
return items[:MAX_STATUS_ENTRIES]
|
|
return items
|
|
|
|
staged_count = len(staged)
|
|
modified_count = len(modified)
|
|
untracked_count = len(untracked)
|
|
changed_count = len({entry["path"] for entry in (*staged, *modified, *untracked)})
|
|
additions = sum(added for added, _ in (*staged_numstat.values(), *modified_numstat.values()))
|
|
deletions = sum(deleted for _, deleted in (*staged_numstat.values(), *modified_numstat.values()))
|
|
|
|
staged = _bounded(staged)
|
|
modified = _bounded(modified)
|
|
untracked = _bounded(untracked)
|
|
|
|
return {
|
|
"counts": {
|
|
"staged": staged_count,
|
|
"modified": modified_count,
|
|
"untracked": untracked_count,
|
|
"changes": changed_count,
|
|
"additions": additions,
|
|
"deletions": deletions,
|
|
},
|
|
"staged": staged,
|
|
"modified": modified,
|
|
"untracked": untracked,
|
|
"truncated": truncated,
|
|
}
|
|
|
|
|
|
def repo_branches(repo: Path) -> list[dict[str, Any]]:
|
|
"""Return branch list with name, upstream, ahead/behind, is_current."""
|
|
current = ""
|
|
try:
|
|
current = _git(repo, "symbolic-ref", "--short", "-q", "HEAD").strip()
|
|
except GitStateError:
|
|
pass
|
|
|
|
# %(upstream:track) yields "[ahead 1, behind 2]", "[gone]", or "".
|
|
fmt = "%(refname:short)%00%(upstream:short)%00%(upstream:track)"
|
|
raw = _git(repo, "for-each-ref", "refs/heads", f"--format={fmt}")
|
|
branches: list[dict[str, Any]] = []
|
|
for line in raw.splitlines():
|
|
if not line:
|
|
continue
|
|
name, upstream, track = line.split("\x00", 2)
|
|
ahead, behind = _parse_track(track)
|
|
branches.append(
|
|
{
|
|
"name": name,
|
|
"upstream": upstream or None,
|
|
"ahead": ahead,
|
|
"behind": behind,
|
|
"is_current": name == current,
|
|
}
|
|
)
|
|
return branches
|
|
|
|
|
|
def _parse_track(track: str) -> tuple[int, int]:
|
|
"""Parse ``[ahead 1, behind 2]`` / ``[gone]`` / ``""`` into (ahead, behind)."""
|
|
if not track or track == "[gone]":
|
|
return 0, 0
|
|
ahead = behind = 0
|
|
for part in re.findall(r"(ahead|behind)\s+(\d+)", track):
|
|
if part[0] == "ahead":
|
|
ahead = int(part[1])
|
|
else:
|
|
behind = int(part[1])
|
|
return ahead, behind
|
|
|
|
|
|
def repo_diff(repo: Path, path: str, kind: str) -> dict[str, Any]:
|
|
"""Return a per-file diff for ``kind`` in (``staged``, ``unstaged``)."""
|
|
if kind not in ("staged", "unstaged"):
|
|
raise GitStateError(f"invalid diff kind: {kind}")
|
|
safe_path = resolve_repo_path(repo, path)
|
|
args = ["diff", "--no-color", "--"]
|
|
if kind == "staged":
|
|
args = ["diff", "--cached", "--no-color", "--"]
|
|
output = _git(repo, *args, safe_path, literal_pathspecs=True)
|
|
truncated = len(output) > MAX_DIFF_BYTES
|
|
if truncated:
|
|
output = output[:MAX_DIFF_BYTES]
|
|
return {"path": safe_path, "kind": kind, "diff": output, "truncated": truncated}
|
|
|
|
|
|
@_serialized_repo_operation
|
|
def read_file(repo: Path, path: str) -> dict[str, Any]:
|
|
"""Read a tracked file's working-tree content. Untracked/binary/missing → error."""
|
|
safe_path = resolve_repo_path(repo, path)
|
|
# Confirm the file is tracked before reading.
|
|
try:
|
|
_git(
|
|
repo,
|
|
"ls-files",
|
|
"--error-unmatch",
|
|
"--",
|
|
safe_path,
|
|
literal_pathspecs=True,
|
|
)
|
|
except GitStateError as exc:
|
|
raise GitStateError(f"file is not tracked: {safe_path}") from exc
|
|
|
|
# Read the working-tree file from disk (not `git show HEAD:`), so a
|
|
# modified-but-uncommitted file returns what is on disk. Read bytes first:
|
|
# binary content dies on the NUL check (before any decode), and non-UTF-8
|
|
# text raises a clear GitStateError instead of an unhandled 500.
|
|
try:
|
|
disk_path = _resolve_repo_disk_path(repo, safe_path, strict=True)
|
|
with disk_path.open("rb") as handle:
|
|
# Re-resolve after opening, then prove the open handle still names
|
|
# that in-repo file. This fails closed if a parent, junction, repo
|
|
# root, or leaf is replaced between validation and open; reads use
|
|
# the already-verified handle after this point.
|
|
revalidated = _resolve_repo_disk_path(repo, safe_path, strict=True)
|
|
if not os.path.samestat(os.fstat(handle.fileno()), revalidated.stat()):
|
|
raise GitStateError(f"path changed during read: {safe_path}")
|
|
raw = handle.read(MAX_FILE_BYTES + 1)
|
|
except OSError as exc:
|
|
raise GitStateError(f"could not read file: {safe_path}") from exc
|
|
|
|
if b"\x00" in raw:
|
|
raise GitStateError(f"binary file is not supported: {safe_path}")
|
|
|
|
truncated = len(raw) > MAX_FILE_BYTES
|
|
if truncated:
|
|
raw = raw[:MAX_FILE_BYTES]
|
|
|
|
try:
|
|
content = raw.decode("utf-8", errors="strict")
|
|
except UnicodeDecodeError as exc:
|
|
raise GitStateError(f"file is not valid UTF-8 text: {safe_path}") from exc
|
|
|
|
return {"path": safe_path, "content": content, "truncated": truncated}
|
|
|
|
|
|
# ── Write (mutation) surface ──────────────────────────────────────────────
|
|
# Every mutation requires the plugin's ``plugin.api.write`` grant (enforced at
|
|
# the router boundary, matching the app's existing grant gating). Destructive
|
|
# operations additionally require a per-use confirmation string echoed by the
|
|
# app/tab. The confirmation values are fixed opaque tokens chosen here; the
|
|
# client shows the human-readable description and sends the token back.
|
|
CONFIRM_DISCARD = "discard"
|
|
CONFIRM_PUSH = "push"
|
|
CONFIRM_DIRTY_CHECKOUT = "checkout-dirty"
|
|
|
|
# Bounds keeping payloads and subprocess argv lists bounded.
|
|
MAX_MUTATION_PATHS = 200
|
|
MAX_COMMIT_MESSAGE = 500
|
|
|
|
# Return-code markers used to classify git failures into the structured error
|
|
# taxonomy the UI renders (non-repo, dirty, conflict, auth, network).
|
|
_DIRTY_MARKERS = (
|
|
"local changes",
|
|
"would be overwritten",
|
|
"cannot pull with rebase",
|
|
"Your local changes to the following files would be overwritten",
|
|
"You have unstaged changes",
|
|
"contains uncommitted changes",
|
|
"working tree contains modifications",
|
|
)
|
|
_CONFLICT_MARKERS = (
|
|
"merge conflict",
|
|
"CONFLICT",
|
|
"Automatic merge failed",
|
|
"fix conflicts",
|
|
"Merge conflict",
|
|
)
|
|
_AUTH_MARKERS = (
|
|
"Authentication failed",
|
|
"could not read Username",
|
|
"Permission denied (publickey)",
|
|
"does not appear to be a git repository",
|
|
"Repository not found",
|
|
"Invalid username or password",
|
|
"authentication failed",
|
|
"could not read Password",
|
|
)
|
|
_NETWORK_MARKERS = (
|
|
"Could not resolve host",
|
|
"Connection timed out",
|
|
"Network is unreachable",
|
|
"Operation timed out",
|
|
"Could not read from remote repository",
|
|
"unable to access",
|
|
"Failed to connect",
|
|
"Name or service not known",
|
|
"getaddrinfo",
|
|
"Temporary failure in name resolution",
|
|
)
|
|
|
|
|
|
def _classify_git_failure(stderr: str) -> str:
|
|
"""Map a git stderr to a stable taxonomy code."""
|
|
for marker in _DIRTY_MARKERS:
|
|
if marker in stderr:
|
|
return "dirty"
|
|
for marker in _CONFLICT_MARKERS:
|
|
if marker in stderr:
|
|
return "conflict"
|
|
for marker in _AUTH_MARKERS:
|
|
if marker in stderr:
|
|
return "auth"
|
|
for marker in _NETWORK_MARKERS:
|
|
if marker in stderr:
|
|
return "network"
|
|
return "invalid-input"
|
|
|
|
|
|
def _require_confirmation(confirmation: str | None, expected: str) -> None:
|
|
"""Enforce the per-use confirmation string for a destructive mutation."""
|
|
if not confirmation:
|
|
raise GitError("this action requires confirmation", code="missing-confirmation")
|
|
if confirmation != expected:
|
|
raise GitError("confirmation did not match", code="wrong-confirmation")
|
|
|
|
|
|
def _validate_paths(paths: list[str] | None) -> list[str]:
|
|
"""Normalize a bounded list of repo-relative paths with traversal checks."""
|
|
if not paths:
|
|
raise GitStateError("paths are required")
|
|
if len(paths) > MAX_MUTATION_PATHS:
|
|
raise GitStateError(f"too many paths (max {MAX_MUTATION_PATHS})")
|
|
return [resolve_repo_path(Path("."), p) for p in paths]
|
|
|
|
|
|
def _is_git_repo(path: Path) -> bool:
|
|
"""True if ``path`` is a git work tree (has a .git entry)."""
|
|
return (path / ".git").exists()
|
|
|
|
|
|
def _run_mutation(
|
|
repo: Path,
|
|
args: list[str],
|
|
*,
|
|
literal_pathspecs: bool = False,
|
|
) -> str:
|
|
"""Run a mutating ``git -C <repo> <args>``; raise a classified GitError.
|
|
|
|
Arg lists only (never shell interpolation); bounded by a timeout.
|
|
"""
|
|
returncode, stdout, stderr_output = _run_git_bounded(
|
|
repo,
|
|
args,
|
|
mutation=True,
|
|
literal_pathspecs=literal_pathspecs,
|
|
)
|
|
if returncode != 0:
|
|
stderr = _safe_git_error(stderr_output or stdout)
|
|
raise GitError(
|
|
f"git {args[0] if args else 'command'} failed for {repo.name}: {stderr}",
|
|
code=_classify_git_failure(stderr),
|
|
)
|
|
return stdout
|
|
|
|
|
|
def _mutate(
|
|
repo: Path,
|
|
args: list[str],
|
|
*,
|
|
literal_pathspecs: bool = False,
|
|
) -> str:
|
|
"""Validate ``repo`` is a real git work tree, then run the mutation."""
|
|
if not _is_git_repo(repo):
|
|
raise GitError(f"not a git repository: {repo.name}", code="non-repo")
|
|
return _run_mutation(repo, args, literal_pathspecs=literal_pathspecs)
|
|
|
|
|
|
def _is_tracked_path(repo: Path, path: str) -> bool:
|
|
"""Classify one literal path without treating Git failures as untracked."""
|
|
returncode, stdout, stderr = _run_git_bounded(
|
|
repo,
|
|
["ls-files", "--error-unmatch", "--", path],
|
|
mutation=False,
|
|
literal_pathspecs=True,
|
|
)
|
|
if returncode == 0:
|
|
return True
|
|
if returncode == 1:
|
|
return False
|
|
raise GitStateError(
|
|
f"git ls-files failed for {repo.name}: {_safe_git_error(stderr or stdout)}"
|
|
)
|
|
|
|
|
|
def _is_dirty(repo: Path) -> bool:
|
|
try:
|
|
return bool(_git(repo, "status", "--porcelain").strip())
|
|
except GitError:
|
|
return False
|
|
|
|
|
|
def _validate_commit_message(message: str) -> str:
|
|
if not isinstance(message, str) or not message.strip():
|
|
raise GitError("commit message must not be empty", code="invalid-input")
|
|
return message.strip()[:MAX_COMMIT_MESSAGE]
|
|
|
|
|
|
def _validate_git_scalar(value: str, label: str, *, allow_empty: bool = False) -> str:
|
|
if not isinstance(value, str):
|
|
raise GitError(f"{label} must be a string", code="invalid-input")
|
|
value = value.strip()
|
|
if not value:
|
|
if allow_empty:
|
|
return ""
|
|
raise GitError(f"{label} is required", code="invalid-input")
|
|
if len(value) > MAX_GIT_SCALAR_LENGTH:
|
|
raise GitError(f"{label} is too long", code="invalid-input")
|
|
if value.startswith("-"):
|
|
raise GitError(f"{label} must not be a git option", code="invalid-input")
|
|
if "\x00" in value or any(ord(char) < 32 for char in value):
|
|
raise GitError(f"{label} contains invalid characters", code="invalid-input")
|
|
return value
|
|
|
|
|
|
def _validate_remote(repo: Path, remote: str) -> str:
|
|
remote = _validate_git_scalar(remote, "remote")
|
|
configured = {line.strip() for line in _git(repo, "remote").splitlines() if line.strip()}
|
|
if remote not in configured:
|
|
raise GitError(f"unknown remote: {remote}", code="invalid-input")
|
|
return remote
|
|
|
|
|
|
def _validate_branch(repo: Path, branch: str, *, allow_empty: bool = False) -> str:
|
|
branch = _validate_git_scalar(branch, "branch", allow_empty=allow_empty)
|
|
if not branch:
|
|
return ""
|
|
try:
|
|
_git(repo, "check-ref-format", "--branch", branch)
|
|
except GitStateError as exc:
|
|
raise GitError(f"invalid branch: {branch}", code="invalid-input") from exc
|
|
return branch
|
|
|
|
|
|
def _validate_revision(repo: Path, ref: str, *, allow_empty: bool = False) -> str:
|
|
ref = _validate_git_scalar(ref, "ref", allow_empty=allow_empty)
|
|
if not ref:
|
|
return ""
|
|
try:
|
|
_git(repo, "rev-parse", "--verify", "--end-of-options", f"{ref}^{{commit}}")
|
|
except GitStateError as exc:
|
|
raise GitError(f"unknown ref: {ref}", code="invalid-input") from exc
|
|
return ref
|
|
|
|
|
|
def _checkout_args(
|
|
repo: Path,
|
|
ref: str,
|
|
*,
|
|
new_branch: str,
|
|
track: bool,
|
|
) -> list[str]:
|
|
new_branch = _validate_branch(repo, new_branch, allow_empty=True)
|
|
ref = _validate_revision(repo, ref, allow_empty=bool(new_branch))
|
|
if track and not ref:
|
|
raise GitError("track requires a source ref", code="invalid-input")
|
|
if new_branch:
|
|
if not ref:
|
|
_validate_revision(repo, "HEAD")
|
|
returncode, _, error = _run_git_bounded(
|
|
repo,
|
|
["show-ref", "--verify", "--quiet", f"refs/heads/{new_branch}"],
|
|
mutation=False,
|
|
)
|
|
if returncode == 0:
|
|
raise GitError(f"branch already exists: {new_branch}", code="invalid-input")
|
|
if returncode != 1:
|
|
raise GitError(
|
|
f"could not validate branch {new_branch}: {_safe_git_error(error)}",
|
|
code="invalid-input",
|
|
)
|
|
args = ["checkout"]
|
|
if track:
|
|
args.append("--track")
|
|
args.extend(["-b", new_branch])
|
|
if ref:
|
|
args.append(ref)
|
|
return args
|
|
args = ["checkout"]
|
|
if track:
|
|
args.append("--track")
|
|
args.append(ref)
|
|
return args
|
|
|
|
|
|
def _fresh_mutation_result(
|
|
repo: Path,
|
|
extra: dict[str, Any] | None = None,
|
|
) -> dict[str, Any]:
|
|
"""Post-mutation snapshot: new HEAD oid + fresh working-tree status."""
|
|
head = ""
|
|
try:
|
|
head = _git(repo, "rev-parse", "HEAD").strip()
|
|
except GitError:
|
|
# Unborn branch — no HEAD yet.
|
|
pass
|
|
result: dict[str, Any] = {"head": head, "status": repo_status(repo)}
|
|
if extra:
|
|
result.update(extra)
|
|
return result
|
|
|
|
|
|
@_serialized_repo_operation
|
|
def stage(repo: Path, paths: list[str]) -> dict[str, Any]:
|
|
"""Stage ``paths`` (repo-relative) and return fresh status."""
|
|
safe = _validate_paths(paths)
|
|
_mutate(repo, ["add", "--"] + safe, literal_pathspecs=True)
|
|
return _fresh_mutation_result(repo)
|
|
|
|
|
|
@_serialized_repo_operation
|
|
def unstage(repo: Path, paths: list[str]) -> dict[str, Any]:
|
|
"""Unstage ``paths`` and return fresh status."""
|
|
safe = _validate_paths(paths)
|
|
_mutate(repo, ["restore", "--staged", "--"] + safe, literal_pathspecs=True)
|
|
return _fresh_mutation_result(repo)
|
|
|
|
|
|
@_serialized_repo_operation
|
|
def discard(
|
|
repo: Path,
|
|
paths: list[str],
|
|
confirmation: str | None,
|
|
delete_untracked: bool = False,
|
|
) -> dict[str, Any]:
|
|
"""Discard local changes to ``paths`` (confirmation required).
|
|
|
|
With ``delete_untracked`` the named untracked files are removed from disk.
|
|
Returns fresh status.
|
|
"""
|
|
_require_confirmation(confirmation, CONFIRM_DISCARD)
|
|
safe = _validate_paths(paths)
|
|
tracked: list[str] = []
|
|
untracked: list[str] = []
|
|
for path in safe:
|
|
if _is_tracked_path(repo, path):
|
|
tracked.append(path)
|
|
elif delete_untracked:
|
|
untracked.append(path)
|
|
# Reject links and directories before asking Git to remove only the exact
|
|
# literal file names. Concurrent same-user filesystem mutation is outside
|
|
# the plugin trust boundary; stationary redirections fail closed here.
|
|
deletable = [_validate_untracked_delete_path(repo, path) for path in untracked]
|
|
if deletable:
|
|
_mutate(repo, ["clean", "-f", "--"] + deletable, literal_pathspecs=True)
|
|
# Revert tracked modifications for the given paths.
|
|
if tracked:
|
|
_mutate(repo, ["checkout", "--"] + tracked, literal_pathspecs=True)
|
|
return _fresh_mutation_result(repo)
|
|
|
|
|
|
@_serialized_repo_operation
|
|
def commit(repo: Path, message: str) -> dict[str, Any]:
|
|
"""Create a commit from the staged index. Empty message is rejected."""
|
|
message = _validate_commit_message(message)
|
|
_mutate(repo, ["commit", "-m", message])
|
|
return _fresh_mutation_result(repo)
|
|
|
|
|
|
@_serialized_repo_operation
|
|
def commit_selected(repo: Path, message: str, paths: list[str]) -> dict[str, Any]:
|
|
"""Commit only the given ``paths`` (staged + modified) under ``message``."""
|
|
message = _validate_commit_message(message)
|
|
safe = _validate_paths(paths)
|
|
_mutate(repo, ["commit", "-m", message, "--"] + safe, literal_pathspecs=True)
|
|
return _fresh_mutation_result(repo)
|
|
|
|
|
|
@_serialized_repo_operation
|
|
def fetch(repo: Path, remote: str = "origin") -> dict[str, Any]:
|
|
"""Fetch from ``remote`` (default origin) and return fresh status/branches."""
|
|
remote = _validate_remote(repo, remote)
|
|
_mutate(repo, ["fetch", "--prune", remote])
|
|
return _fresh_mutation_result(repo, {"branches": repo_branches(repo)})
|
|
|
|
|
|
@_serialized_repo_operation
|
|
def pull(repo: Path, remote: str = "origin", branch: str = "") -> dict[str, Any]:
|
|
"""Pull from ``remote``/``branch`` (defaults: origin + current branch).
|
|
|
|
Pull never clobbers local work: a tree git refuses to fast-forward without
|
|
discarding local changes surfaces as a structured ``dirty`` GitError.
|
|
"""
|
|
remote = _validate_remote(repo, remote)
|
|
branch = _validate_branch(repo, branch, allow_empty=True)
|
|
args = ["pull", "--ff-only", remote]
|
|
if branch:
|
|
args.append(branch)
|
|
try:
|
|
_mutate(repo, args)
|
|
except GitError as exc:
|
|
if exc.code == "dirty":
|
|
raise GitError(
|
|
"Pull would overwrite local changes — commit or discard them first.",
|
|
code="dirty",
|
|
) from exc
|
|
raise
|
|
return _fresh_mutation_result(repo)
|
|
|
|
|
|
@_serialized_repo_operation
|
|
def push(
|
|
repo: Path,
|
|
remote: str = "origin",
|
|
branch: str = "",
|
|
confirmation: str | None = None,
|
|
) -> dict[str, Any]:
|
|
"""Push to ``remote``/``branch``. Confirmation is required.
|
|
|
|
Auth/network failures surface as classified GitErrors; fresh status/branches
|
|
are returned so the UI can reflect ahead/behind after a successful push.
|
|
"""
|
|
_require_confirmation(confirmation, CONFIRM_PUSH)
|
|
remote = _validate_remote(repo, remote)
|
|
branch = _validate_branch(repo, branch, allow_empty=True)
|
|
args = ["push", remote]
|
|
if branch:
|
|
args.append(branch)
|
|
_mutate(repo, args)
|
|
return _fresh_mutation_result(repo, {"branches": repo_branches(repo)})
|
|
|
|
|
|
@_serialized_repo_operation
|
|
def checkout(
|
|
repo: Path,
|
|
ref: str,
|
|
confirmation: str | None = None,
|
|
new_branch: str = "",
|
|
track: bool = False,
|
|
) -> dict[str, Any]:
|
|
"""Switch to ``ref`` (optionally creating ``new_branch``, optionally --track).
|
|
|
|
A dirty tree switch requires confirmation. Git still refuses to overwrite
|
|
conflicting local changes, so there is no data-loss path.
|
|
"""
|
|
args = _checkout_args(repo, ref, new_branch=new_branch, track=track)
|
|
|
|
if _is_dirty(repo) and not new_branch:
|
|
_require_confirmation(confirmation, CONFIRM_DIRTY_CHECKOUT)
|
|
_mutate(repo, args)
|
|
return _fresh_mutation_result(repo, {"branches": repo_branches(repo)})
|
|
|
|
|
|
def _staged_diff(repo: Path, paths: list[str] | None) -> tuple[str, bool]:
|
|
"""Return the bounded staged diff for ``paths`` (or all staged when None).
|
|
|
|
Returns ``(diff_text, truncated)``. The diff is capped at MAX_DIFF_BYTES so
|
|
the LLM prompt stays bounded. Untracked/modified-but-unstaged content is
|
|
never included: only what is staged is eligible for a commit message.
|
|
"""
|
|
args = ["diff", "--cached", "--no-color"]
|
|
if paths:
|
|
args.append("--")
|
|
args.extend(paths)
|
|
output = _git(repo, *args, literal_pathspecs=bool(paths))
|
|
truncated = len(output) > MAX_DIFF_BYTES
|
|
if truncated:
|
|
output = output[:MAX_DIFF_BYTES]
|
|
return output, truncated
|
|
|
|
|
|
def _llm_call(messages: list[dict[str, str]]):
|
|
"""Call the agent's in-process LLM on a chat ``messages`` list.
|
|
|
|
Reuses the plugin's existing upstream plumbing: ``agent.auxiliary_client``
|
|
(the same in-process helper the upstream tools use). The import is deferred
|
|
so the module stays importable in test envs without the agent package; the
|
|
message-generation tests monkeypatch ``_llm_call``/``_llm_extract``.
|
|
"""
|
|
client = _resolve_llm()
|
|
return client(messages=messages)
|
|
|
|
|
|
def _llm_extract(response: Any) -> str:
|
|
from agent.auxiliary_client import extract_content_or_reasoning
|
|
|
|
return extract_content_or_reasoning(response)
|
|
|
|
|
|
def _resolve_llm():
|
|
"""Resolve the deferred async model client callable (raises if absent)."""
|
|
from agent.auxiliary_client import async_call_llm
|
|
|
|
return async_call_llm
|
|
|
|
|
|
async def _generate_message(diff: str) -> dict[str, Any]:
|
|
"""Build the LLM prompt for a bounded staged diff and return the suggestion.
|
|
|
|
A missing model or failed call degrades to an empty message with a
|
|
``model unavailable`` notice — never an exception/500.
|
|
"""
|
|
if not diff.strip():
|
|
return {"message": "", "notice": "nothing staged"}
|
|
messages = [
|
|
{
|
|
"role": "user",
|
|
"content": (
|
|
"Write a conventional commit message for the staged diff below. "
|
|
"Return ONLY the message: a subject line under 72 characters in "
|
|
"the form 'type: summary', then a blank line, then an optional "
|
|
"body describing what and why. Do not wrap in quotes.\n\n"
|
|
f"<staged diff>\n{diff}\n</staged diff>"
|
|
),
|
|
}
|
|
]
|
|
try:
|
|
response = await _llm_call(messages=messages)
|
|
text = _llm_extract(response).strip()
|
|
except Exception as exc: # noqa: BLE001
|
|
logger.info("commit message generation unavailable: %s", exc)
|
|
return {"message": "", "notice": "model unavailable for commit messages"}
|
|
if not text:
|
|
return {"message": "", "notice": "model returned no suggestion"}
|
|
return {"message": text, "notice": ""}
|
|
|
|
|
|
async def commit_message(repo: Path) -> dict[str, Any]:
|
|
"""Generate a conventional-style commit-message suggestion from the staged diff.
|
|
|
|
Empty staged diff → ``{"message": "", "notice": "nothing staged"}`` WITHOUT
|
|
calling the LLM. A missing model or failed call degrades to an empty message
|
|
with a ``model unavailable`` notice — never an exception/500. Only staged
|
|
content is ever sent.
|
|
"""
|
|
diff, _ = _staged_diff(repo, None)
|
|
return await _generate_message(diff)
|
|
|
|
|
|
async def commit_message_selected(
|
|
repo: Path, paths: list[str]
|
|
) -> dict[str, Any]:
|
|
"""Generate a commit-message suggestion from the staged diff of ``paths``.
|
|
|
|
Only the given paths' staged content is considered; a path with nothing
|
|
staged contributes nothing. Empty staged diff → no LLM call.
|
|
"""
|
|
safe = _validate_paths(paths)
|
|
diff, _ = _staged_diff(repo, safe)
|
|
return await _generate_message(diff)
|
|
|
|
|
|
@_serialized_repo_operation
|
|
def stash_checkout(
|
|
repo: Path,
|
|
ref: str,
|
|
new_branch: str = "",
|
|
track: bool = False,
|
|
) -> dict[str, Any]:
|
|
"""Checkout that auto-stashes a dirty tree first.
|
|
|
|
Dirty tree: ``git stash push -m "git-state: <ref>"`` then checkout; returns
|
|
``{stashed: true, stash_message}`` so the UI can surface the stash. Clean
|
|
tree: plain checkout with ``{stashed: false}``. No confirmation is required
|
|
because a stash is recoverable (``git stash``), unlike discard — documented
|
|
in the endpoint and UI. Git still refuses to overwrite conflicting changes,
|
|
so there is no data-loss path. ``new_branch``/``track`` mirror the plain
|
|
checkout surface.
|
|
"""
|
|
args = _checkout_args(repo, ref, new_branch=new_branch, track=track)
|
|
|
|
stashed = False
|
|
stash_message = ""
|
|
stash_oid = ""
|
|
if _is_dirty(repo):
|
|
stash_message = f"git-state: {ref or new_branch}"
|
|
_mutate(repo, ["stash", "push", "--include-untracked", "-m", stash_message])
|
|
stash_oid = _git(repo, "rev-parse", "--verify", "refs/stash").strip()
|
|
stashed = True
|
|
|
|
try:
|
|
_mutate(repo, args)
|
|
except GitError as checkout_error:
|
|
if not stashed:
|
|
raise
|
|
try:
|
|
_mutate(repo, ["stash", "apply", "--index", stash_oid])
|
|
except GitError as restore_error:
|
|
raise GitError(
|
|
f"{checkout_error}; changes remain in stash {stash_oid}; "
|
|
f"automatic restore failed: {restore_error}",
|
|
code=checkout_error.code,
|
|
) from checkout_error
|
|
raise GitError(
|
|
f"{checkout_error}; working changes were restored and remain backed up "
|
|
f"in stash {stash_oid}",
|
|
code=checkout_error.code,
|
|
) from checkout_error
|
|
return {
|
|
**{"stashed": stashed, "stash_message": stash_message},
|
|
**_fresh_mutation_result(repo, {"branches": repo_branches(repo)}),
|
|
}
|
|
|
|
|
|
def repo_remotes(repo: Path) -> list[dict[str, str]]:
|
|
"""Return remote names + URLs with embedded userinfo scrubbed."""
|
|
raw = _git(repo, "remote", "-v")
|
|
remotes: list[dict[str, str]] = []
|
|
seen: set[str] = set()
|
|
for line in raw.splitlines():
|
|
parts = line.split()
|
|
if len(parts) < 2:
|
|
continue
|
|
name, url = parts[0], parts[1]
|
|
if name in seen:
|
|
continue
|
|
seen.add(name)
|
|
remotes.append({"name": name, "url": _scrub_url(url)})
|
|
return remotes
|
|
|
|
|
|
def _scrub_url(url: str) -> str:
|
|
"""Strip userinfo (``user:pass@`` / ``user@``) from a remote URL."""
|
|
scrubbed = _USERINFO_RE.sub(r"\1", url)
|
|
# scp-style: user@host:path — drop the user but keep the host:path.
|
|
scrubbed = _SSH_USERINFO_RE.sub(r"\2:", scrubbed)
|
|
return scrubbed
|
|
|
|
|
|
def build_git_document(base: Path) -> dict[str, Any]:
|
|
"""Build the declarative Git page document served at ``mobile/pages/git``.
|
|
|
|
The document is a read-only snapshot: it lists scanned repositories as
|
|
literal text and carries no filesystem paths (per the android-plugins.md
|
|
document contract). Interactive repo → status → diff/file browsing is
|
|
provided by the dedicated Android Compose screen, which calls the
|
|
``/git/*`` API endpoints directly; this document is the manifest-declared
|
|
fallback surface.
|
|
"""
|
|
repos = scan_repos(base)
|
|
children: list[dict[str, Any]] = []
|
|
if not base.is_dir():
|
|
children.append(
|
|
{
|
|
"type": "text",
|
|
"id": "notice",
|
|
"text": {
|
|
"type": "literal",
|
|
"value": "Configured Git base path was not found. Check the plugin setting.",
|
|
},
|
|
}
|
|
)
|
|
elif not repos:
|
|
children.append(
|
|
{
|
|
"type": "text",
|
|
"id": "empty",
|
|
"text": {
|
|
"type": "literal",
|
|
"value": "No repositories found under the configured base path.",
|
|
},
|
|
}
|
|
)
|
|
else:
|
|
for repo in repos:
|
|
dirty = " (dirty)" if repo["dirty"] else ""
|
|
children.append(
|
|
{
|
|
"type": "text",
|
|
"id": f"repo-{repo['id']}",
|
|
"text": {
|
|
"type": "literal",
|
|
"value": f"{repo['name']} — {repo['current_branch'] or 'detached'}{dirty}",
|
|
},
|
|
}
|
|
)
|
|
|
|
return {
|
|
"schemaVersion": 1,
|
|
"pages": [
|
|
{
|
|
"id": "git",
|
|
"title": {"type": "literal", "value": "Git"},
|
|
"content": {
|
|
"type": "group",
|
|
"id": "git-root",
|
|
"direction": "column",
|
|
"children": children,
|
|
},
|
|
}
|
|
],
|
|
}
|
|
|
|
|
|
__all__ = [
|
|
"DEFAULT_BASE_PATH",
|
|
"GIT_TIMEOUT_SECONDS",
|
|
"GitStateError",
|
|
"MAX_DIFF_BYTES",
|
|
"MAX_FILE_BYTES",
|
|
"MAX_STATUS_ENTRIES",
|
|
"base_path",
|
|
"build_git_document",
|
|
"commit_message",
|
|
"commit_message_selected",
|
|
"read_file",
|
|
"repo_branches",
|
|
"repo_diff",
|
|
"repo_id",
|
|
"repo_remotes",
|
|
"repo_status",
|
|
"resolve_repo",
|
|
"resolve_repo_path",
|
|
"scan_repos",
|
|
"stash_checkout",
|
|
]
|