diff --git a/src/reyna_cli/cli.py b/src/reyna_cli/cli.py index 85dad27..ebd2174 100644 --- a/src/reyna_cli/cli.py +++ b/src/reyna_cli/cli.py @@ -22,6 +22,7 @@ from reyna_cli.apple_llm_client import ( 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.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.mcp import MCPClient 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.") zoom_app = typer.Typer(help="Zoom direct REST API 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.") 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.") @@ -79,6 +81,7 @@ app.add_typer(immich_app, name="immich") app.add_typer(mongo_app, name="mongo") app.add_typer(zoom_app, name="zoom") app.add_typer(email_app, name="email") +app.add_typer(gitea_app, name="gitea") app.add_typer(deco_app, name="deco") macmini_app.add_typer(macmini_calendar_app, name="calendar") 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) +@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: device = get_device(name) if not device: diff --git a/src/reyna_cli/gitea_direct.py b/src/reyna_cli/gitea_direct.py new file mode 100644 index 0000000..c51438e --- /dev/null +++ b/src/reyna_cli/gitea_direct.py @@ -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}") diff --git a/tests/test_gitea_direct.py b/tests/test_gitea_direct.py new file mode 100644 index 0000000..4d1e36c --- /dev/null +++ b/tests/test_gitea_direct.py @@ -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"}