feat: restore Gitea CLI access

This commit is contained in:
Adolfo Reyna
2026-08-16 08:40:47 -04:00
parent 032bc3a580
commit bdf3b55a98
3 changed files with 297 additions and 0 deletions
+39
View File
@@ -22,6 +22,7 @@ from reyna_cli.apple_llm_client import (
from reyna_cli.deco_direct import DecoDirectClient from reyna_cli.deco_direct import DecoDirectClient
from reyna_cli.desktop_service import SERVICE_NAME, service_action, service_status, unit_content, unit_path from reyna_cli.desktop_service import SERVICE_NAME, service_action, service_status, unit_content, unit_path
from reyna_cli.email_local import ThunderbirdEmailClient, email_tool_specs from reyna_cli.email_local import ThunderbirdEmailClient, email_tool_specs
from reyna_cli.gitea_direct import GiteaClient, load_credentials
from reyna_cli.immich import ImmichClient from reyna_cli.immich import ImmichClient
from reyna_cli.mcp import MCPClient from reyna_cli.mcp import MCPClient
from reyna_cli.mongo_direct import MongoDirectClient from reyna_cli.mongo_direct import MongoDirectClient
@@ -51,6 +52,7 @@ immich_app = typer.Typer(help="Immich direct REST API commands.")
mongo_app = typer.Typer(help="MongoDB direct driver commands.") mongo_app = typer.Typer(help="MongoDB direct driver commands.")
zoom_app = typer.Typer(help="Zoom direct REST API commands.") zoom_app = typer.Typer(help="Zoom direct REST API commands.")
email_app = typer.Typer(help="Read-only local Thunderbird email commands.") email_app = typer.Typer(help="Read-only local Thunderbird email commands.")
gitea_app = typer.Typer(help="Direct Gitea API and credential diagnostics (never prints tokens).")
deco_app = typer.Typer(help="TP-Link Deco direct router commands.") deco_app = typer.Typer(help="TP-Link Deco direct router commands.")
macmini_app = typer.Typer(help="Mac mini MCP tools (Calendar, Contacts, Reminders, Deco).") macmini_app = typer.Typer(help="Mac mini MCP tools (Calendar, Contacts, Reminders, Deco).")
remarkable_app = typer.Typer(help="Paper Pro discovery, local cache, and macOS listener service.") remarkable_app = typer.Typer(help="Paper Pro discovery, local cache, and macOS listener service.")
@@ -79,6 +81,7 @@ app.add_typer(immich_app, name="immich")
app.add_typer(mongo_app, name="mongo") app.add_typer(mongo_app, name="mongo")
app.add_typer(zoom_app, name="zoom") app.add_typer(zoom_app, name="zoom")
app.add_typer(email_app, name="email") app.add_typer(email_app, name="email")
app.add_typer(gitea_app, name="gitea")
app.add_typer(deco_app, name="deco") app.add_typer(deco_app, name="deco")
macmini_app.add_typer(macmini_calendar_app, name="calendar") macmini_app.add_typer(macmini_calendar_app, name="calendar")
macmini_app.add_typer(macmini_contacts_app, name="contacts") macmini_app.add_typer(macmini_contacts_app, name="contacts")
@@ -170,6 +173,42 @@ def fail(message: str, json_output: bool = False, **extra: Any) -> None:
raise typer.Exit(1) raise typer.Exit(1)
@gitea_app.command("credential")
def gitea_credential(json_output: bool = typer.Option(False, "--json")):
"""Report whether a Gitea credential is configured, without displaying it."""
emit({"ok": True, **load_credentials().public_dict()}, json_output)
@gitea_app.command("access")
def gitea_access(json_output: bool = typer.Option(False, "--json")):
"""Read the authenticated Gitea user endpoint."""
try:
emit({"ok": True, "user": GiteaClient().access()}, json_output)
except Exception as exc:
fail(str(exc), json_output)
@gitea_app.command("repos")
def gitea_repos(
limit: int = typer.Option(50, min=1, max=100),
json_output: bool = typer.Option(False, "--json"),
):
"""List repositories when the configured token has repository-list scope."""
try:
emit({"ok": True, "repos": GiteaClient().repos(limit)}, json_output)
except Exception as exc:
fail(str(exc), json_output)
@gitea_app.command("repo")
def gitea_repo(name: str, json_output: bool = typer.Option(False, "--json")):
"""Read one repository and its permissions, e.g. adolforeyna/reyna-cli."""
try:
emit({"ok": True, "repo": GiteaClient().repo(name)}, json_output)
except Exception as exc:
fail(str(exc), json_output)
def get_required_device(name: str, json_output: bool = False) -> Device: def get_required_device(name: str, json_output: bool = False) -> Device:
device = get_device(name) device = get_device(name)
if not device: if not device:
+174
View File
@@ -0,0 +1,174 @@
from __future__ import annotations
import base64
import json
import os
import subprocess
from dataclasses import dataclass
from pathlib import Path
from typing import Any
from urllib.parse import unquote, urlsplit
import httpx
from reyna_cli.env import load_hermes_env
DEFAULT_GITEA_URL = "https://git.reynafamily.com"
SENSITIVE_KEYS = {"token", "access_token", "password", "authorization", "refresh_token"}
@dataclass(frozen=True)
class GiteaCredentials:
username: str
token: str
source: str
base_url: str
def public_dict(self) -> dict[str, Any]:
parsed = urlsplit(self.base_url)
return {
"configured": bool(self.token) or self.source == "Pi 5 git credential helper",
"host": parsed.netloc,
"source": self.source,
}
def gitea_url() -> str:
load_hermes_env()
return os.getenv("REYNA_GITEA_URL") or os.getenv("GITEA_URL") or DEFAULT_GITEA_URL
def _credential_from_file(path: Path, host: str) -> tuple[str, str] | None:
try:
lines = path.read_text(encoding="utf-8", errors="ignore").splitlines()
except OSError:
return None
for line in lines:
parsed = urlsplit(line.strip())
if parsed.hostname == host and parsed.username and parsed.password:
return unquote(parsed.username), unquote(parsed.password)
return None
def _credential_from_git_helper(host: str) -> tuple[str, str] | None:
try:
result = subprocess.run(
["git", "credential", "fill"],
input=f"protocol=https\nhost={host}\n\n",
text=True,
capture_output=True,
timeout=10,
check=False,
)
except (OSError, subprocess.TimeoutExpired):
return None
if result.returncode:
return None
values = dict(line.split("=", 1) for line in result.stdout.splitlines() if "=" in line)
username, password = values.get("username", ""), values.get("password", "")
return (username, password) if username and password else None
def load_credentials() -> GiteaCredentials:
load_hermes_env()
base_url = gitea_url().rstrip("/")
host = urlsplit(base_url).hostname or ""
token = os.getenv("REYNA_GITEA_TOKEN") or os.getenv("GITEA_TOKEN")
if token:
return GiteaCredentials(os.getenv("REYNA_GITEA_USERNAME", "adolforeyna"), token, "REYNA_GITEA_TOKEN", base_url)
credentials_path = Path(os.getenv("GIT_CREDENTIALS_FILE", Path.home() / ".git-credentials"))
found = _credential_from_file(credentials_path, host)
if found:
return GiteaCredentials(*found, "GIT_CREDENTIALS_FILE", base_url)
found = _credential_from_git_helper(host)
if found:
return GiteaCredentials(*found, "git credential helper", base_url)
return GiteaCredentials("", "", "Pi 5 git credential helper", base_url)
def redact_sensitive(value: Any) -> Any:
if isinstance(value, dict):
return {key: "[REDACTED]" if key.casefold() in SENSITIVE_KEYS else redact_sensitive(item) for key, item in value.items()}
if isinstance(value, list):
return [redact_sensitive(item) for item in value]
return value
class GiteaClient:
def __init__(self, credentials: GiteaCredentials | None = None, timeout: float = 20.0, transport: httpx.BaseTransport | None = None):
self.credentials = credentials or load_credentials()
self.remote_credential_helper = self.credentials.source == "Pi 5 git credential helper"
if not self.credentials.token and not self.remote_credential_helper:
raise RuntimeError("No Gitea credential is available. Set REYNA_GITEA_TOKEN or configure a supported Git credential source.")
self.client = httpx.Client(
base_url=self.credentials.base_url,
timeout=timeout,
transport=transport,
headers={"Authorization": f"token {self.credentials.token}", "Accept": "application/json"},
)
def _remote_get(self, path: str) -> Any:
"""Use Pi 5's existing Git credential helper without returning its token."""
script = r'''import json, subprocess, sys, urllib.request
from urllib.parse import urlsplit
base_url, path = sys.argv[1:]
host = urlsplit(base_url).hostname
result = subprocess.run(["git", "credential", "fill"], input=f"protocol=https\nhost={host}\n\n", text=True, capture_output=True, check=False)
values = dict(line.split("=", 1) for line in result.stdout.splitlines() if "=" in line)
token = values.get("password", "")
if result.returncode or not token:
raise SystemExit("Pi 5 Git credential helper did not provide a Gitea credential")
request = urllib.request.Request(base_url.rstrip("/") + path, headers={"Authorization": "token " + token, "Accept": "application/json"})
with urllib.request.urlopen(request, timeout=20) as response:
payload = json.load(response)
def redact(value):
if isinstance(value, dict):
return {key: "[REDACTED]" if key.casefold() in {"token", "access_token", "password", "authorization", "refresh_token"} else redact(item) for key, item in value.items()}
if isinstance(value, list):
return [redact(item) for item in value]
return value
print(json.dumps(redact(payload)))'''
helper_host = os.getenv("REYNA_GITEA_CREDENTIAL_HOST", "pi5")
encoded_script = base64.b64encode(script.encode("utf-8")).decode("ascii")
remote_command = (
"python3 -c \"import base64;exec(base64.b64decode('"
+ encoded_script
+ "'))\" "
+ self.credentials.base_url
+ " "
+ path
)
result = subprocess.run(
["ssh", "-o", "BatchMode=yes", helper_host, remote_command],
text=True,
capture_output=True,
timeout=30,
check=False,
)
if result.returncode:
raise RuntimeError("Pi 5 Gitea credential helper request failed.")
try:
return json.loads(result.stdout)
except json.JSONDecodeError as exc:
raise RuntimeError("Pi 5 Gitea credential helper returned invalid JSON.") from exc
def _get(self, path: str) -> Any:
if self.remote_credential_helper:
return self._remote_get(path)
response = self.client.get(path)
response.raise_for_status()
return redact_sensitive(response.json())
def access(self) -> Any:
return self._get("/api/v1/user")
def repos(self, limit: int = 50) -> Any:
return self._get("/api/v1/user/repos?limit=" + str(limit))
def repo(self, name: str) -> Any:
owner, separator, repo = name.partition("/")
if not separator or not owner or not repo:
raise ValueError("Repository must be in owner/name form.")
return self._get(f"/api/v1/repos/{owner}/{repo}")
+84
View File
@@ -0,0 +1,84 @@
from __future__ import annotations
from pathlib import Path
import httpx
from reyna_cli.gitea_direct import GiteaClient, GiteaCredentials, load_credentials, redact_sensitive
def test_load_credentials_prefers_environment_without_exposing_token(monkeypatch):
monkeypatch.setenv("REYNA_GITEA_TOKEN", "test-token-value")
monkeypatch.setenv("REYNA_GITEA_URL", "https://git.example.test/")
credentials = load_credentials()
assert credentials.token == "test-token-value"
assert credentials.source == "REYNA_GITEA_TOKEN"
assert credentials.public_dict() == {
"configured": True,
"host": "git.example.test",
"source": "REYNA_GITEA_TOKEN",
}
assert "token" not in credentials.public_dict()
def test_load_credentials_reads_git_credentials_file(monkeypatch, tmp_path: Path):
credentials_file = tmp_path / "credentials"
credentials_file.write_text(
"https://adolforeyna:stored-secret@git.reynafamily.com\n",
encoding="utf-8",
)
monkeypatch.delenv("REYNA_GITEA_TOKEN", raising=False)
monkeypatch.delenv("GITEA_TOKEN", raising=False)
monkeypatch.setenv("GIT_CREDENTIALS_FILE", str(credentials_file))
credentials = load_credentials()
assert credentials.username == "adolforeyna"
assert credentials.token == "stored-secret"
assert credentials.source == "GIT_CREDENTIALS_FILE"
assert "stored-secret" not in str(credentials.public_dict())
def test_repo_response_is_redacted_and_requests_authenticated_endpoint():
requests: list[httpx.Request] = []
def handler(request: httpx.Request) -> httpx.Response:
requests.append(request)
return httpx.Response(
200,
json={
"full_name": "adolforeyna/reyna-cli",
"permissions": {"push": True},
"token": "response-secret",
},
)
client = GiteaClient(
GiteaCredentials("adolforeyna", "request-secret", "test", "https://git.example.test"),
transport=httpx.MockTransport(handler),
)
result = client.repo("adolforeyna/reyna-cli")
assert requests[0].url.path == "/api/v1/repos/adolforeyna/reyna-cli"
assert requests[0].headers["authorization"] == "token request-secret"
assert result["permissions"]["push"] is True
assert result["token"] == "[REDACTED]"
assert "response-secret" not in str(result)
def test_remote_credential_helper_can_make_a_safe_api_request(monkeypatch):
credentials = GiteaCredentials("", "", "Pi 5 git credential helper", "https://git.example.test")
assert credentials.public_dict()["configured"] is True
client = GiteaClient(credentials)
monkeypatch.setattr(client, "_remote_get", lambda path: {"full_name": "adolforeyna/reyna-cli"})
assert client.repo("adolforeyna/reyna-cli") == {"full_name": "adolforeyna/reyna-cli"}
def test_redact_sensitive_handles_nested_api_values():
assert redact_sensitive(
{"password": "one", "nested": [{"access_token": "two"}], "name": "safe"}
) == {"password": "[REDACTED]", "nested": [{"access_token": "[REDACTED]"}], "name": "safe"}