From 3117ca5ae4809f9c3618bb04de719f9a2c46401d Mon Sep 17 00:00:00 2001 From: csh Date: Wed, 22 Jul 2026 17:28:22 +0800 Subject: [PATCH] :sparkles: feat(gitea-fix-ci): add authenticated log collector --- skills/gitea-fix-ci/SKILL.md | 265 ++++--- skills/gitea-fix-ci/scripts/fetch_ci_logs.py | 699 +++++++++++++++++++ test/test_gitea_fix_ci_skill.py | 274 ++++++++ 3 files changed, 1135 insertions(+), 103 deletions(-) create mode 100644 skills/gitea-fix-ci/scripts/fetch_ci_logs.py create mode 100644 test/test_gitea_fix_ci_skill.py diff --git a/skills/gitea-fix-ci/SKILL.md b/skills/gitea-fix-ci/SKILL.md index d43bddcb..a57b6fef 100644 --- a/skills/gitea-fix-ci/SKILL.md +++ b/skills/gitea-fix-ci/SKILL.md @@ -1,151 +1,210 @@ --- name: gitea-fix-ci -description: Use when a user asks to debug or fix failing Gitea Actions, Gitea PR checks, or CI workflow runs for a Gitea-hosted repository. +description: "Use when a user asks to debug or fix failing Gitea Actions, Gitea PR checks, or CI workflow runs for a Gitea-hosted repository. Also triggers on Chinese phrasing such as CI 挂了/流水线红了/构建失败/工作流失败." --- # Gitea Fix CI -## Overview +## 概述 -Diagnose failing Gitea Actions from the pull request or workflow run, extract the -smallest useful failure context, then propose a fix plan before changing code. -Core principle: CI logs are evidence; do not guess from the red status alone. +从 pull request 或 workflow run 诊断失败的 Gitea Actions，提取最小可用的失败上下文， +然后在改代码之前先给出修复计划。核心原则：CI 日志是证据，不要仅凭红色状态臆测原因。 -This is not a standalone executor. It guides use of `tea`, local `git`, and the -Gitea API from the current workspace. +本 skill 不是独立执行器。它指导你在当前工作区中使用随附的取证脚本 +`scripts/fetch_ci_logs.py`、`tea`、本地 `git` 和 Gitea API。 -## When to Use +取证阶段（step 2-4）优先使用随附脚本，把「探测版本 → 找失败 run → 列 job → +下载失败 job 日志」这段易记错 API 路径的逻辑交给它。脚本仅取证到日志，分类、 +修复计划与改代码仍由你按本文档执行。将 `` 替换为包含本 `SKILL.md` +的目录： -- A Gitea-hosted repository has failing Gitea Actions or PR checks -- The user asks to inspect a failed workflow run, job, or CI status -- The user asks to fix CI after a push, branch update, or pull request update -- Local tests pass but remote Gitea Actions fail +```bash +# 方式一（优先）：通过环境变量提供 API token +export GITEA_TOKEN= +python /scripts/fetch_ci_logs.py version # 探测实例版本并确认连通性 +python /scripts/fetch_ci_logs.py runs --status failure --branch +python /scripts/fetch_ci_logs.py jobs # 标注 failure 的 job +python /scripts/fetch_ci_logs.py logs # 日志存临时文件并回显尾部 -## When Not to Use +# 方式二：显式复用当前仓库的 Git HTTP 凭据 +python /scripts/fetch_ci_logs.py --use-git-credential version +python /scripts/fetch_ci_logs.py --use-git-credential runs --status failure +``` -- The repository is not hosted on Gitea or Forgejo-compatible infrastructure -- The failure belongs to an external CI provider and only links out from Gitea -- The user only wants a local test or lint run -- Credentials, tokens, or network access are unavailable and the user has not - provided the failing log text +`base-url`、owner 和 repo 默认从 git remote 推导。认证规则如下： -## Inputs +- 设置了 `GITEA_TOKEN` 时始终优先使用它，即使同时传入 `--use-git-credential`。 +- 只有显式传入 `--use-git-credential` 才会非交互调用 `git credential fill`，并将 + 当前仓库对应的用户名和密码用于 HTTP Basic 认证；脚本不会静默读取 Git 凭据。 +- 未设置 token 且未传入该参数时按匿名方式访问，私有仓库通常会返回 401。 +- 所有携带凭据的请求必须使用 HTTPS；认证 header 只发送到配置的 Gitea 同源地址， + 跨源请求和跨源重定向会被拒绝。凭据不会出现在命令参数、日志或异常文本中。 -- Repository path, defaulting to the current workspace -- Gitea base URL and repository owner/name, from `git remote -v` when possible -- Pull request number, branch, commit SHA, or workflow run ID -- Authentication method: `tea` login profile, `GITEA_TOKEN` for API requests, - or an existing git credential for the Gitea web fallback -- Any pasted CI log if remote access is unavailable +脚本不可用（未装 Python、旧版 Gitea 无 job-log API、脱离 Gitea 环境）时，回退到 +下文的手动 `tea`/API/web 路径。 -## Procedure +## 适用场景 -1. **Baseline local state** +- Gitea 托管的仓库出现失败的 Gitea Actions 或 PR checks +- 用户要求检查某个失败的 workflow run、job 或 CI 状态 +- 用户要求在 push、更新分支或更新 pull request 之后修复 CI +- 本地测试通过，但远端 Gitea Actions 失败 - - Record `git status --short`, current branch, and latest commit SHA. - - Identify the Gitea remote URL and owner/repo. - - Do not modify files while gathering CI evidence. +## 不适用场景 -2. **Verify Gitea access** +- 仓库并非托管在 Gitea 或 Forgejo 兼容的基础设施上 +- 失败属于外部 CI 服务，只是从 Gitea 链接出去 +- 用户只想在本地跑测试或 lint +- 没有凭据、token 或网络访问权限，且用户也未提供失败的日志文本 - - Prefer `tea` if it is installed and authenticated: +## 输入 + +- 仓库路径，默认为当前工作区 +- Gitea base URL 和仓库 owner/name，尽量从 `git remote -v` 获取 +- Pull request 编号、分支、commit SHA 或 workflow run ID +- 认证方式：`tea` 登录 profile、用于 API 请求的 `GITEA_TOKEN`，或通过 + `--use-git-credential` 显式读取的当前仓库 Git HTTP 凭据 +- 若无法远端访问，则由用户粘贴的 CI 日志 + +## 流程 + +1. **基线本地状态** + + - 记录 `git status --short`、当前分支和最新 commit SHA。 + - 从 `git remote -v` 识别 remote URL 和 owner/repo。 + - 继续之前先确认 remote 是 Gitea/Forgejo：对照已知的 Gitea 实例核对 host， + 或探测 `/api/v1/version`。若 remote 是 GitHub、GitLab 或其他服务，按"不适用场景"停止。 + - 收集 CI 证据期间不要修改文件。 + +2. **验证 Gitea 访问** + + - 优先运行 `fetch_ci_logs.py version` 探测实例版本并确认脚本能连通实例。脚本从 + remote 推导 base URL，认证按 `GITEA_TOKEN` → 显式 `--use-git-credential` → + 匿名的顺序选择。注意：`version` 只读 `/api/v1/version`，成功仅代表连通与鉴权 + 可用，**不代表 Actions API 一定可用**。Actions 端点是否存在需在后续 + `runs`/`jobs` 步骤中实际验证；旧版 Gitea（见下）会在此才暴露 404。 + - 脚本不可用时，回退到手动方式。优先使用已安装并已认证的 `tea`： - `tea login list` + - 使用前先运行 `tea actions --help` 确认已安装的 `tea` 版本存在 `actions` + 子命令；并非所有版本都带这些子命令。 - `tea actions runs list --status failure --branch ` - `tea actions runs view ` - `tea actions runs logs --job ` - `tea pulls view ` - - If `tea` is unavailable or lacks an Actions command for the installed - version, use the Gitea API directly. - - Check `/api/v1/version` and `/swagger.v1.json` when API behavior is - unclear. Gitea 1.21 exposes Actions pages but not workflow run/job/log API - endpoints. - - Never print tokens. Pass API tokens through environment variables. + - 若 `tea` 不可用，或已安装版本缺少 Actions 命令，则直接使用 Gitea API。 + - 当 API 行为不明确时，检查 `/api/v1/version` 和 `/swagger.v1.json`。较旧的 + Gitea（1.21 及更早）在 web UI 中提供 Actions 页面，但不提供 workflow + run/job/log 的 API 端点；应对照上报的版本确认可用性，而不是想当然。 + - 绝不打印 token、用户名或密码。API token 只通过环境变量传递；Git 凭据只通过 + `git credential fill` 的标准输入/输出在脚本进程内传递，不放入 argv。 -3. **Find failing workflow runs** +3. **定位失败的 workflow run** - - For a known run ID, fetch that run directly. - - Otherwise list recent workflow runs filtered by branch, event, status, or - commit SHA. - - API patterns: + - 若入口是 PR 而非 run ID，先把 PR 解析成 run：取 PR 的 head 分支与 head SHA + （`fetch_ci_logs.py` 的 `runs` 支持 `--branch`/`--sha`；手动则 + `GET /api/v1/repos///pulls/` 取 `head.ref`/`head.sha`）， + 再按该 SHA 过滤 runs。PR 页面的 checks 列表可能聚合多个 workflow，逐一定位到 + 具体失败 run，不要假设只有一个。 + - 优先用脚本：`fetch_ci_logs.py runs --status failure --branch ` + （也支持 `--sha`/`--event`/`--limit`），已知 run ID 时用 + `fetch_ci_logs.py jobs ` 直接列出各 job 并标注失败项。 + - 脚本不可用时回退到手动方式。已知 run ID 时，直接获取该 run。 + - 否则按分支、事件、状态或 commit SHA 过滤，列出最近的 workflow runs。 + - API 模式： - `GET /api/v1/repos///actions/runs` - `GET /api/v1/repos///actions/runs/` - `GET /api/v1/repos///actions/runs//jobs` - - Web fallback for older Gitea: + - 较旧 Gitea 的 web 回退： - `GET ///actions` - - Parse `///actions/runs/` links and status labels. - - Treat `failure`, `cancelled`, and missing required checks differently. - Cancelled jobs may require rerun or queue investigation rather than code - changes. + - 解析 `///actions/runs/` 链接和状态标签。 + - 区别对待 `failure`、`cancelled` 和缺失的必需 checks。被取消（cancelled）的 + job 可能需要重跑或排查队列，而非改代码。 -4. **Fetch job logs** +4. **获取 job 日志** - - For each failed job, download the job logs: + - 优先用脚本：`fetch_ci_logs.py logs `。它把日志存到受跟踪源码路径 + 之外的临时文件、只回显尾部若干行，并打印临时文件路径供进一步查看 + （`--tail N` 调整行数，`--json` 输出结构化结果）。 + - 脚本不可用时回退到手动方式。对每个失败的 job，下载其日志： - `GET /api/v1/repos///actions/jobs//logs` - - On Gitea 1.21 web fallback, download logs by UI job index: + - 在较旧 Gitea 的 web 回退（无 job-log API）下，按 UI 的 job index 下载日志： - `GET ///actions/runs//jobs//logs` - - First open the run page and read each job's status; take the index of a - job whose status is `failure`, not index `0` by default. The failing job - is rarely the first one, so a blind index `0` usually returns a passing - job's log. Only fall back to scanning indices when the run page does not - expose per-job status. - - Save large logs to a temporary file outside tracked source paths. - - Extract the first actionable error block, surrounding command, job name, - workflow name, run URL, branch, and SHA. - - If logs are missing, report that explicitly instead of inventing causes. + - 先打开 run 页面读取每个 job 的状态；取状态为 `failure` 的 job 的 index， + 不要默认用 index `0`。失败的 job 很少是第一个，盲目用 index `0` 通常会返回 + 某个通过的 job 的日志。只有当 run 页面不暴露每个 job 的状态时，才回退到逐一 + 扫描 index。 + - 大日志保存到受跟踪源码路径之外的临时文件。 + - 提取首个可操作的错误块、其上下文命令、job 名、workflow 名、run URL、分支和 SHA。 + - 若日志缺失，明确报告，而不是编造原因。 -5. **Classify the failure** +5. **分类失败** - - Code/test failure: failing assertion, compile error, lint error, type error - - Environment failure: missing secret, runner image, dependency install, - network, cache, permission, or service startup - - Workflow failure: invalid YAML, unsupported syntax, wrong trigger, bad path, - wrong branch/ref assumption - - Infrastructure failure: offline runner, stuck queue, cancelled run, timeout + - 代码/测试失败：断言失败、编译错误、lint 错误、类型错误 + - 环境失败：缺失 secret、runner 镜像、依赖安装、网络、缓存、权限或服务启动 + - 工作流失败：YAML 非法、语法不支持、触发条件错误、路径错误、分支/ref 假设错误 + - 基础设施失败：runner 离线、队列卡住、run 被取消、超时 -6. **Create a fix plan** +6. **制定修复计划** - - Summarize the failure evidence with exact job/run identifiers. - - Propose the smallest code or workflow change that matches the evidence. - - Include local verification commands and the remote recheck path. - - Do not implement before the user approves the fix plan. + - 用确切的 job/run 标识总结失败证据。 + - 提出与证据匹配的最小代码或工作流改动。 + - 包含本地验证命令和远端复检路径。 + - 在用户批准修复计划之前不要实施。 -7. **Implement after approval** +7. **批准后实施** - - Apply only the approved fix. - - Run the local command that most closely reproduces the failed job. - - If the failure is workflow-only, validate the workflow file syntax and any - referenced paths or scripts. + - 只应用已批准的修复。 + - 运行最接近复现该失败 job 的本地命令。 + - 若失败仅涉及工作流，验证工作流文件语法及其引用的路径或脚本。 -8. **Recheck** +8. **复检** - - Tell the user what must be pushed or rerun in Gitea. - - If permitted, use the Gitea API to inspect the rerun status. - - Final output must distinguish local verification from remote CI status. + - 告诉用户需要在 Gitea 中 push 或重跑什么。 + - 若获准，使用 Gitea API 检查重跑状态。 + - 最终输出必须区分本地验证与远端 CI 状态。 -## Output Contract +## 输出约定 -- `Target:` repo, branch/SHA, PR or run ID -- `Failed CI:` workflow, job, status, run URL or API path -- `Evidence:` concise log snippet and classification -- `Plan:` proposed fix, local verification, remote recheck -- `Changes:` files changed after approval -- `Result:` local checks run and remaining remote status +- `Target:` 仓库、分支/SHA、PR 或 run ID +- `Failed CI:` workflow、job、状态、run URL 或 API 路径 +- `Evidence:` 精简的日志片段与分类 +- `Plan:` 提出的修复、本地验证、远端复检 +- `Changes:` 批准后改动的文件 +- `Result:` 已运行的本地检查与剩余的远端状态 -## Success Criteria +## 成功标准 -- Failure analysis is based on Gitea Actions run/job data or pasted logs -- The fix plan names the exact workflow run or job it addresses -- No code or workflow edits happen before plan approval -- Verification distinguishes local commands from remote Gitea Actions results -- Tokens and private log content are not echoed unnecessarily +- 失败分析基于 Gitea Actions 的 run/job 数据或粘贴的日志 +- 修复计划指明其针对的确切 workflow run 或 job +- 计划批准前不发生任何代码或工作流改动 +- 验证区分本地命令与远端 Gitea Actions 结果 +- 不回显 token；私有日志只保留可操作的片段，不整段外泄 -## Failure Handling +## 危险信号（Red Flags） -- If authentication fails, ask the user to authenticate `tea` or provide a token - through the environment; do not request secrets in chat -- If the Gitea version lacks Actions API endpoints, ask for the relevant log text - or a browser-copied job log -- If an external CI provider owns the failing check, report the external URL and - stop at evidence collection -- If the failure is infrastructure-only, recommend rerun/runner investigation - instead of editing code +出现以下情况说明流程走偏，停下纠正而非继续： + +- **凭红色状态臆测原因**：还没下载 job 日志就断言失败原因或动手改代码。 +- **误读双字段结果**：把 `status`（生命周期）当成结果判定。Gitea Actions 沿用 + GitHub 兼容的双字段模型——`status=completed` 只表示跑完了，真正的成败在 + `conclusion`（`failure`/`success`/`cancelled`）。判定失败必须看有效结果， + 而非 `status=completed` 就当通过。 +- **盲取 job index `0`**：在 web 回退下不看每个 job 状态就用 index `0` 下载日志； + 失败的 job 很少是第一个，通常会误取到某个通过 job 的日志。 +- **把 `version` 成功当作 Actions API 可用**：`version` 只探连通与鉴权，Actions + 端点可能仍返回 404（旧版 Gitea）。 +- **把 `cancelled`/基础设施问题当代码 bug 修**：被取消、runner 离线、队列卡住、 + 超时应重跑或排查环境，不是改代码。 +- **回显 token 或整段私有日志**：只保留可操作的最小片段。 +- **静默读取或降级传输凭据**：Git 凭据必须由 `--use-git-credential` 显式启用； + token 和 Git 凭据都不得通过 HTTP 或跨源重定向发送。 + +## 失败处理 + +- 401 且未使用认证时，通过环境提供 `GITEA_TOKEN` 或显式传入 + `--use-git-credential`；401 且已使用认证时检查凭据是否有效，不要在对话中索要 secret +- 403 表示所选凭据缺少仓库或 Actions 读取权限；补足权限，而不是切换到匿名访问 +- 404 优先核对 owner/repo、端点和 Gitea 版本；旧版本缺少 Actions API 时使用 web 回退 +- 若 Gitea 版本缺少 Actions API 端点，请用户提供相关日志文本或从浏览器复制的 job 日志 +- 若失败的 check 归属外部 CI 服务，报告外部 URL 并在证据收集处停止 +- 若失败仅为基础设施问题，建议重跑/排查 runner，而非改代码 diff --git a/skills/gitea-fix-ci/scripts/fetch_ci_logs.py b/skills/gitea-fix-ci/scripts/fetch_ci_logs.py new file mode 100644 index 00000000..b04aa729 --- /dev/null +++ b/skills/gitea-fix-ci/scripts/fetch_ci_logs.py @@ -0,0 +1,699 @@ +#!/usr/bin/env python3 +"""Collect Gitea Actions CI evidence: version, runs, jobs, and job logs. + +This is an evidence-collection helper for the gitea-fix-ci skill. It automates +the programmatic parts of the skill's Procedure (probe version, list failing +runs, list jobs, download a failing job's log) so the agent does not hand-build +API paths or blindly pick job index 0. It stops at evidence: classification, +fix plans, and code edits stay with the agent. + +Scope decisions (see SKILL.md): +- Evidence only. No classification, no fix, no edits. +- Gitea Actions REST API only. No legacy web-scraping fallback; when the API is + absent the tool points the agent back to SKILL.md's manual/web path. +- Auth via GITEA_TOKEN (preferred) or an explicitly requested `git credential` + lookup; base URL/owner/repo are derived from `git remote -v`. Secrets are + never placed on argv, logs, or exception text. + +Zero third-party dependencies (stdlib urllib only). +""" + +from __future__ import annotations + +import argparse +import base64 +import json +import os +import re +import subprocess +import sys +import tempfile +import urllib.error +import urllib.parse +import urllib.request +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any + +TOKEN_ENV = "GITEA_TOKEN" +DEFAULT_REMOTE = "origin" +DEFAULT_LOG_TAIL = 40 +REQUEST_TIMEOUT = 30 # seconds; a hung Gitea/proxy must not block the session +CREDENTIAL_TIMEOUT = 10 # seconds; helpers must not block on an interactive UI +USER_AGENT = "gitea-fix-ci/fetch_ci_logs" + +# https://host/owner/repo(.git) or git@host:owner/repo(.git) or +# ssh://git@host[:port]/owner/repo(.git) +_HTTP_REMOTE_RE = re.compile( + r"^(?Phttps?)://(?:[^@/]+@)?(?P[^/]+)/" + r"(?P[^/]+)/(?P[^/]+?)(?:\.git)?/?$" +) +_SCP_REMOTE_RE = re.compile( + r"^(?:ssh://)?(?:[^@]+@)?(?P[^:/]+)(?::\d+)?[:/]" + r"(?P[^/]+)/(?P[^/]+?)(?:\.git)?/?$" +) + + +class ConfigError(ValueError): + """Missing or malformed configuration (remote, token, arguments).""" + + +class ApiError(RuntimeError): + """The Gitea API returned an error or unexpected payload.""" + + +def _eprint(*args: object) -> None: + print(*args, file=sys.stderr) + + +def _origin(url: str) -> tuple[str, str, int]: + """Return a normalized (scheme, hostname, effective port) tuple.""" + try: + parsed = urllib.parse.urlsplit(url) + hostname = parsed.hostname + port = parsed.port + except ValueError: + raise ConfigError("invalid Gitea URL") from None + + if parsed.scheme.lower() not in {"http", "https"} or not parsed.netloc: + raise ConfigError("Gitea URL must use http:// or https://") + if parsed.username is not None or parsed.password is not None: + raise ConfigError("Gitea URL must not contain embedded credentials") + if not hostname: + raise ConfigError("Gitea URL must include a host") + + scheme = parsed.scheme.lower() + effective_port = port if port is not None else (443 if scheme == "https" else 80) + return scheme, hostname.rstrip(".").lower(), effective_port + + +def _normalize_base_url(value: str) -> str: + """Validate and normalize the configured API base URL.""" + if not isinstance(value, str) or not value.strip(): + raise ConfigError("Gitea base URL is required") + value = value.strip().rstrip("/") + try: + parsed = urllib.parse.urlsplit(value) + except ValueError: + raise ConfigError("invalid Gitea base URL") from None + if parsed.query or parsed.fragment: + raise ConfigError("Gitea base URL must not contain a query or fragment") + _origin(value) + return value + + +@dataclass(frozen=True) +class RepoTarget: + base_url: str # e.g. https://git.example.com + owner: str + repo: str + + def __post_init__(self) -> None: + object.__setattr__(self, "base_url", _normalize_base_url(self.base_url)) + if not self.owner or not self.repo: + raise ConfigError("Gitea owner and repository are required") + + @property + def api_root(self) -> str: + return f"{self.base_url}/api/v1" + + def repo_path(self, suffix: str) -> str: + owner = urllib.parse.quote(self.owner, safe="") + repo = urllib.parse.quote(self.repo, safe="") + return f"{self.api_root}/repos/{owner}/{repo}{suffix}" + + @property + def origin(self) -> tuple[str, str, int]: + return _origin(self.base_url) + + @property + def credential_host(self) -> str: + parsed = urllib.parse.urlsplit(self.base_url) + hostname = parsed.hostname or "" + if ":" in hostname and not hostname.startswith("["): + hostname = f"[{hostname}]" + if parsed.port is not None: + hostname = f"{hostname}:{parsed.port}" + return hostname + + @property + def credential_path(self) -> str: + base_path = urllib.parse.urlsplit(self.base_url).path.strip("/") + owner = urllib.parse.quote(self.owner, safe="") + repo = urllib.parse.quote(self.repo, safe="") + repository_path = f"{owner}/{repo}.git" + return f"{base_path}/{repository_path}" if base_path else repository_path + + +@dataclass(frozen=True) +class ApiAuth: + """An authorization header whose secret is deliberately absent from repr.""" + + source: str + authorization: str = field(repr=False) + + def __post_init__(self) -> None: + scheme, separator, credential = self.authorization.partition(" ") + has_control = any( + ord(character) < 32 or ord(character) == 127 + for character in self.authorization + ) + try: + self.authorization.encode("ascii") + except UnicodeEncodeError: + raise ConfigError("invalid authorization value") from None + if ( + not self.source + or not separator + or not scheme + or not credential + or has_control + ): + raise ConfigError("invalid authorization value") + + +def _require_https(target: RepoTarget) -> None: + if target.origin[0] != "https": + raise ConfigError("credentials require an HTTPS Gitea base URL") + + +def _run_git(args: list[str]) -> str: + try: + result = subprocess.run( + ["git", *args], + capture_output=True, + text=True, + check=False, + ) + except OSError as exc: + raise ConfigError(f"cannot run git {' '.join(args)}: {exc}") from exc + if result.returncode != 0: + detail = result.stderr.strip() + suffix = f": {detail}" if detail else "" + raise ConfigError(f"git {' '.join(args)} failed{suffix}") + return result.stdout.strip() + + +def _remote_url(remote: str) -> str: + url = _run_git(["remote", "get-url", remote]) + if not url: + raise ConfigError(f"remote '{remote}' has no URL") + return url + + +def parse_remote_url(url: str) -> tuple[str, str, str]: + """Return (base_url, owner, repo) parsed from a git remote URL. + + Only http(s) remotes yield a usable API base URL. SSH remotes give host and + path but no scheme, so we assume https for the API base. + """ + http_match = _HTTP_REMOTE_RE.match(url) + if http_match: + base = f"{http_match.group('scheme')}://{http_match.group('host')}" + return base, http_match.group("owner"), http_match.group("repo") + + scp_match = _SCP_REMOTE_RE.match(url) + if scp_match: + # No scheme in an SSH remote; the API is reached over https by default. + base = f"https://{scp_match.group('host')}" + return base, scp_match.group("owner"), scp_match.group("repo") + + raise ConfigError( + "cannot parse owner/repo from the configured git remote. Gitea remotes " + "are expected as //; for nested or " + "non-standard paths, pass --base-url/--owner/--repo explicitly." + ) + + +def resolve_target(args: argparse.Namespace) -> RepoTarget: + base_url = args.base_url + owner = args.owner + repo = args.repo + if not (base_url and owner and repo): + url = _remote_url(args.remote) + parsed_base, parsed_owner, parsed_repo = parse_remote_url(url) + base_url = base_url or parsed_base + owner = owner or parsed_owner + repo = repo or parsed_repo + return RepoTarget(base_url=base_url, owner=owner, repo=repo) + + +def _token() -> str | None: + token = os.getenv(TOKEN_ENV) + if token and token.strip(): + return token.strip() + return None + + +def _git_credential_auth(target: RepoTarget) -> ApiAuth: + """Resolve repository-scoped HTTP Basic credentials without prompting.""" + _require_https(target) + credential_query = ( + "protocol=https\n" + f"host={target.credential_host}\n" + f"path={target.credential_path}\n\n" + ) + helper_env = os.environ.copy() + helper_env["GIT_TERMINAL_PROMPT"] = "0" + helper_env["GCM_INTERACTIVE"] = "Never" + + try: + result = subprocess.run( + ["git", "credential", "fill"], + input=credential_query, + capture_output=True, + text=True, + check=False, + timeout=CREDENTIAL_TIMEOUT, + env=helper_env, + ) + except subprocess.TimeoutExpired: + raise ConfigError("git credential lookup timed out") from None + except OSError: + raise ConfigError("cannot run git credential lookup") from None + + if result.returncode != 0: + # stdout/stderr may contain credentials or helper-specific secret data. + raise ConfigError("git credential lookup failed or requires interaction") + + fields: dict[str, str] = {} + for line in result.stdout.splitlines(): + key, separator, value = line.partition("=") + if separator: + fields[key] = value + + username = fields.get("username", "") + password = fields.get("password", "") + if not username or not password or ":" in username: + raise ConfigError("git credential lookup returned unusable credentials") + + encoded = base64.b64encode(f"{username}:{password}".encode("utf-8")).decode("ascii") + return ApiAuth(source="git credential", authorization=f"Basic {encoded}") + + +def resolve_auth(args: argparse.Namespace, target: RepoTarget) -> ApiAuth | None: + """Resolve authentication once, with environment tokens taking precedence.""" + token = _token() + if token: + _require_https(target) + return ApiAuth(source=TOKEN_ENV, authorization=f"token {token}") + if getattr(args, "use_git_credential", False): + return _git_credential_auth(target) + return None + + +class SameOriginRedirectHandler(urllib.request.HTTPRedirectHandler): + """Follow redirects only while they stay on the configured Gitea origin.""" + + def __init__(self, target: RepoTarget) -> None: + super().__init__() + self._origin = target.origin + + def redirect_request( + self, + req: urllib.request.Request, + fp: Any, + code: int, + msg: str, + headers: Any, + newurl: str, + ) -> urllib.request.Request | None: + absolute_url = urllib.parse.urljoin(req.full_url, newurl) + try: + redirect_origin = _origin(absolute_url) + except ConfigError: + raise ConfigError("refusing an invalid Gitea redirect") from None + if redirect_origin != self._origin: + raise ConfigError("refusing a redirect outside the configured origin") + return super().redirect_request(req, fp, code, msg, headers, absolute_url) + + +class ApiClient: + """Small, origin-bound client for read-only Gitea API requests.""" + + def __init__( + self, + target: RepoTarget, + *, + auth: ApiAuth | None = None, + opener: Any | None = None, + ) -> None: + if auth: + _require_https(target) + self.target = target + self.auth = auth + self._opener = opener or urllib.request.build_opener( + SameOriginRedirectHandler(target) + ) + + def request(self, url: str, *, accept_json: bool = True) -> tuple[int, bytes, str]: + """Perform an origin-bound GET and return status, body, content type.""" + try: + request_origin = _origin(url) + except ConfigError: + raise ConfigError("refusing an invalid Gitea API URL") from None + if request_origin != self.target.origin: + raise ConfigError("refusing a request outside the configured origin") + + headers = {"User-Agent": USER_AGENT} + if accept_json: + headers["Accept"] = "application/json" + if self.auth: + headers["Authorization"] = self.auth.authorization + + request = urllib.request.Request(url, headers=headers, method="GET") + try: + with self._opener.open(request, timeout=REQUEST_TIMEOUT) as response: + body = response.read() + content_type = response.headers.get("Content-Type", "") + return response.status, body, content_type + except urllib.error.HTTPError as exc: + body = exc.read() if hasattr(exc, "read") else b"" + content_type = exc.headers.get("Content-Type", "") if exc.headers else "" + return exc.code, body, content_type + except urllib.error.URLError as exc: + detail = str(exc.reason) + if self.auth: + authorization = self.auth.authorization + secret = authorization.partition(" ")[2] + detail = detail.replace(authorization, "") + if secret: + detail = detail.replace(secret, "") + raise ApiError(f"request to Gitea failed: {detail}") from None + + def get_json(self, url: str) -> Any: + status, body, _ = self.request(url, accept_json=True) + if status != 200: + raise ApiError(_explain_status(status, self.target, self.auth)) + try: + return json.loads(body.decode("utf-8")) + except (UnicodeError, json.JSONDecodeError) as exc: + raise ApiError(f"cannot parse JSON from {url}: {exc}") from None + + +def _explain_status( + status: int, target: RepoTarget, auth: ApiAuth | None = None +) -> str: + if status == 401: + if auth is None: + return ( + f"authentication required (HTTP 401). Set {TOKEN_ENV} or rerun " + "with --use-git-credential; do not paste credentials into chat." + ) + return ( + f"authentication rejected (HTTP 401) for {auth.source}. Check that " + "the credential is valid for this Gitea instance and repository." + ) + if status == 403: + if auth is None: + return ( + f"permission denied (HTTP 403) without authentication. Set " + f"{TOKEN_ENV} or rerun with --use-git-credential." + ) + return ( + f"permission denied (HTTP 403) using {auth.source}. Grant repository " + "and Actions read permission to the selected credential." + ) + if status == 404: + return ( + "endpoint not found (HTTP 404). Check the repository coordinates and " + "Gitea version; older Gitea (1.21 and earlier) lacks the Actions " + "run/job/log API. Fall back to the web UI or pasted logs per SKILL.md." + ) + return f"unexpected HTTP {status} from {target.base_url}" + + +def cmd_version( + args: argparse.Namespace, + target: RepoTarget, + client: ApiClient, +) -> int: + payload = client.get_json(f"{target.api_root}/version") + version = payload.get("version") if isinstance(payload, dict) else None + if not version: + raise ApiError("version endpoint returned no 'version' field") + if args.json: + print(json.dumps({"version": version}, ensure_ascii=False)) + else: + print(f"gitea version: {version}") + print(f"instance: {target.base_url}") + return 0 + + +def _as_list(payload: Any, key: str) -> list[dict[str, Any]]: + """Gitea may wrap collections in an object or return a bare list.""" + if isinstance(payload, list): + return [item for item in payload if isinstance(item, dict)] + if isinstance(payload, dict): + inner = payload.get(key) + if isinstance(inner, list): + return [item for item in inner if isinstance(item, dict)] + return [] + + +def _outcome(item: dict[str, Any]) -> str: + """Resolve the effective result of a run/job. + + Gitea's Actions objects follow the GitHub-compatible two-field model: a + lifecycle `status` (queued/in_progress/completed) plus a terminal + `conclusion` (success/failure/cancelled/...). A completed job reports + `status=completed, conclusion=failure`, so a naive `status or conclusion` + would stop at "completed" and miss the failure. Prefer `conclusion`; fall + back to `status` only when there is no conclusion yet. + """ + conclusion = (item.get("conclusion") or "").strip() + status = (item.get("status") or "").strip() + return (conclusion or status).lower() + + +def _run_row(run: dict[str, Any]) -> dict[str, Any]: + return { + "id": run.get("id"), + "name": run.get("name") or run.get("workflow_id") or "", + "outcome": _outcome(run), + "event": run.get("event") or "", + "branch": run.get("head_branch") or run.get("branch") or "", + "sha": (run.get("head_sha") or run.get("commit_sha") or "")[:12], + "url": run.get("html_url") or run.get("url") or "", + } + + +def cmd_runs( + args: argparse.Namespace, + target: RepoTarget, + client: ApiClient, +) -> int: + query: dict[str, str] = {} + if args.branch: + query["branch"] = args.branch + if args.event: + query["event"] = args.event + if args.status: + query["status"] = args.status + if args.sha: + query["head_sha"] = args.sha + if args.limit: + query["limit"] = str(args.limit) + suffix = "/actions/runs" + if query: + suffix += "?" + urllib.parse.urlencode(query) + payload = client.get_json(target.repo_path(suffix)) + runs = [_run_row(run) for run in _as_list(payload, "workflow_runs")] + if args.sha: + runs = [row for row in runs if row["sha"].startswith(args.sha[:12])] + runs = runs[: args.limit] if args.limit else runs + + if args.json: + print(json.dumps(runs, ensure_ascii=False, indent=2)) + return 0 + if not runs: + print("no matching workflow runs") + return 1 + print(f"{len(runs)} run(s):") + for row in runs: + print( + f" run {row['id']} [{row['outcome']}] {row['name']} " + f"{row['event']} {row['branch']} {row['sha']}" + ) + if row["url"]: + print(f" {row['url']}") + return 0 + + +def _job_row(job: dict[str, Any], index: int) -> dict[str, Any]: + return { + "index": index, + "id": job.get("id"), + "name": job.get("name") or "", + "outcome": _outcome(job), + } + + +def _is_failed(outcome: str) -> bool: + return outcome.lower() in {"failure", "failed", "error"} + + +def cmd_jobs( + args: argparse.Namespace, + target: RepoTarget, + client: ApiClient, +) -> int: + payload = client.get_json(target.repo_path(f"/actions/runs/{args.run_id}/jobs")) + jobs = [_job_row(job, index) for index, job in enumerate(_as_list(payload, "jobs"))] + failed = [job for job in jobs if _is_failed(job["outcome"])] + + if args.json: + print( + json.dumps( + {"jobs": jobs, "failed_job_ids": [j["id"] for j in failed]}, + ensure_ascii=False, + indent=2, + ) + ) + return 0 + + if not jobs: + print(f"run {args.run_id} has no jobs (or endpoint unavailable)") + return 1 + print(f"run {args.run_id} jobs:") + for job in jobs: + marker = " <== FAILED" if _is_failed(job["outcome"]) else "" + print( + f" job {job['id']} index={job['index']} " + f"[{job['outcome']}] {job['name']}{marker}" + ) + if failed: + ids = ", ".join(str(job["id"]) for job in failed) + print(f"failed job id(s): {ids}") + print("fetch a failed job's log with: logs ") + else: + print("no failed jobs on this run") + return 0 + + +def cmd_logs( + args: argparse.Namespace, + target: RepoTarget, + client: ApiClient, +) -> int: + status, body, _ = client.request( + target.repo_path(f"/actions/jobs/{args.job_id}/logs"), + accept_json=False, + ) + if status != 200: + raise ApiError(_explain_status(status, target, client.auth)) + + text = body.decode("utf-8", errors="replace") + lines = text.splitlines() + + if args.out: + out_path = Path(args.out).expanduser() + out_path.write_text(text, encoding="utf-8") + else: + handle = tempfile.NamedTemporaryFile( + prefix=f"gitea-job-{args.job_id}-", + suffix=".log", + delete=False, + mode="w", + encoding="utf-8", + ) + handle.write(text) + handle.close() + out_path = Path(handle.name) + + tail = args.tail if args.tail is not None else DEFAULT_LOG_TAIL + tail_lines = lines[-tail:] if tail > 0 else lines + + if args.json: + print( + json.dumps( + { + "job_id": args.job_id, + "log_path": str(out_path), + "total_lines": len(lines), + "tail": tail_lines, + }, + ensure_ascii=False, + indent=2, + ) + ) + return 0 + + print(f"job {args.job_id} log saved to: {out_path}") + print(f"total lines: {len(lines)} (showing last {len(tail_lines)})") + print("-" * 60) + for line in tail_lines: + print(line) + return 0 + + +def _build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser( + description="Collect Gitea Actions CI evidence (version/runs/jobs/logs).", + ) + parser.add_argument("--remote", default=DEFAULT_REMOTE, help="git remote name") + parser.add_argument("--base-url", help="Gitea base URL override") + parser.add_argument("--owner", help="repository owner override") + parser.add_argument("--repo", help="repository name override") + parser.add_argument( + "--use-git-credential", + action="store_true", + help=( + "explicitly use repository-scoped git credentials over HTTPS " + f"when {TOKEN_ENV} is unset" + ), + ) + parser.add_argument( + "--json", action="store_true", help="emit machine-readable JSON" + ) + sub = parser.add_subparsers(dest="command", required=True) + + sub.add_parser("version", help="probe /api/v1/version") + + runs = sub.add_parser("runs", help="list workflow runs") + runs.add_argument("--branch") + runs.add_argument("--event") + runs.add_argument("--status", default="failure") + runs.add_argument("--sha") + runs.add_argument("--limit", type=int, default=20) + + jobs = sub.add_parser("jobs", help="list jobs of a run, flag failed ones") + jobs.add_argument("run_id") + + logs = sub.add_parser("logs", help="download a job log, print its tail") + logs.add_argument("job_id") + logs.add_argument("--out", help="write full log here instead of a temp file") + logs.add_argument( + "--tail", type=int, help=f"tail lines (default {DEFAULT_LOG_TAIL})" + ) + + return parser + + +_COMMANDS = { + "version": cmd_version, + "runs": cmd_runs, + "jobs": cmd_jobs, + "logs": cmd_logs, +} + + +def main(argv: list[str] | None = None) -> int: + if hasattr(sys.stdout, "reconfigure"): + sys.stdout.reconfigure(encoding="utf-8") + args = _build_parser().parse_args(argv) + try: + target = resolve_target(args) + auth = resolve_auth(args, target) + client = ApiClient(target, auth=auth) + return _COMMANDS[args.command](args, target, client) + except ConfigError as exc: + _eprint(f"ERROR: {exc}") + return 2 + except ApiError as exc: + _eprint(f"ERROR: {exc}") + return 2 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/test/test_gitea_fix_ci_skill.py b/test/test_gitea_fix_ci_skill.py new file mode 100644 index 00000000..6d20a743 --- /dev/null +++ b/test/test_gitea_fix_ci_skill.py @@ -0,0 +1,274 @@ +from __future__ import annotations + +import base64 +import importlib.util +import os +import subprocess +import sys +import unittest +import urllib.request +from pathlib import Path +from types import SimpleNamespace +from unittest import mock + +ROOT = Path(__file__).resolve().parents[1] +SKILL_ROOT = ROOT / "skills" / "gitea-fix-ci" +SCRIPT = SKILL_ROOT / "scripts" / "fetch_ci_logs.py" + + +def load_module(): + spec = importlib.util.spec_from_file_location("fetch_ci_logs", SCRIPT) + if spec is None or spec.loader is None: + raise RuntimeError(f"cannot load {SCRIPT}") + module = importlib.util.module_from_spec(spec) + sys.modules[spec.name] = module + spec.loader.exec_module(module) + return module + + +class FakeResponse: + status = 200 + headers = {"Content-Type": "application/json"} + + def __enter__(self): + return self + + def __exit__(self, *_args): + return False + + def read(self) -> bytes: + return b'{"ok": true}' + + +class RecordingOpener: + def __init__(self) -> None: + self.requests = [] + + def open(self, request, *, timeout): + self.requests.append((request, timeout)) + return FakeResponse() + + +class GiteaFixCiSkillTests(unittest.TestCase): + @classmethod + def setUpClass(cls) -> None: + cls.module = load_module() + + def setUp(self) -> None: + self.target = self.module.RepoTarget( + base_url="https://git.example.test", + owner="team", + repo="project", + ) + + def test_token_auth_takes_precedence_without_reading_git_credentials(self): + args = SimpleNamespace(use_git_credential=True) + with mock.patch.dict(os.environ, {"GITEA_TOKEN": "token-secret"}, clear=True): + with mock.patch.object( + self.module, + "_git_credential_auth", + side_effect=AssertionError("git credential must not run"), + ): + auth = self.module.resolve_auth(args, self.target) + + self.assertEqual(auth.source, "GITEA_TOKEN") + self.assertEqual(auth.authorization, "token token-secret") + self.assertNotIn("token-secret", repr(auth)) + + def test_git_credential_auth_is_explicit_scoped_and_non_interactive(self): + credential = subprocess.CompletedProcess( + args=["git", "credential", "fill"], + returncode=0, + stdout=( + "protocol=https\n" + "host=git.example.test\n" + "username=ci-user\n" + "password=basic-secret\n" + ), + stderr="", + ) + args = SimpleNamespace(use_git_credential=True) + with mock.patch.dict(os.environ, {}, clear=True): + with mock.patch.object( + self.module.subprocess, "run", return_value=credential + ) as run: + auth = self.module.resolve_auth(args, self.target) + + self.assertEqual(auth.source, "git credential") + expected = base64.b64encode(b"ci-user:basic-secret").decode("ascii") + self.assertEqual(auth.authorization, f"Basic {expected}") + self.assertNotIn("basic-secret", repr(auth)) + + call = run.call_args + self.assertEqual(call.args[0], ["git", "credential", "fill"]) + self.assertIn("protocol=https\n", call.kwargs["input"]) + self.assertIn("host=git.example.test\n", call.kwargs["input"]) + self.assertIn("path=team/project.git\n", call.kwargs["input"]) + self.assertEqual(call.kwargs["env"]["GIT_TERMINAL_PROMPT"], "0") + self.assertEqual(call.kwargs["env"]["GCM_INTERACTIVE"], "Never") + + def test_git_credential_scope_includes_gitea_base_path(self): + target = self.module.RepoTarget( + base_url="https://git.example.test/gitea", + owner="team", + repo="project", + ) + self.assertEqual(target.credential_path, "gitea/team/project.git") + + def test_git_credentials_are_not_read_without_explicit_flag(self): + args = SimpleNamespace(use_git_credential=False) + with mock.patch.dict(os.environ, {}, clear=True): + with mock.patch.object( + self.module.subprocess, + "run", + side_effect=AssertionError("git credential must not run"), + ): + auth = self.module.resolve_auth(args, self.target) + self.assertIsNone(auth) + + def test_all_credentials_require_https(self): + target = self.module.RepoTarget( + base_url="http://git.example.test", + owner="team", + repo="project", + ) + with mock.patch.dict(os.environ, {"GITEA_TOKEN": "secret"}, clear=True): + with self.assertRaisesRegex(self.module.ConfigError, "HTTPS"): + self.module.resolve_auth( + SimpleNamespace(use_git_credential=False), target + ) + + with mock.patch.dict(os.environ, {}, clear=True): + with self.assertRaisesRegex(self.module.ConfigError, "HTTPS"): + self.module.resolve_auth( + SimpleNamespace(use_git_credential=True), target + ) + + auth = self.module.ApiAuth( + source="GITEA_TOKEN", + authorization="token sensitive-value", + ) + with self.assertRaisesRegex(self.module.ConfigError, "HTTPS"): + self.module.ApiClient(target, auth=auth, opener=RecordingOpener()) + + def test_invalid_token_header_is_rejected_without_echoing_secret(self): + secret = "token-secret\ninjected-header" + with mock.patch.dict(os.environ, {"GITEA_TOKEN": secret}, clear=True): + with self.assertRaises(self.module.ConfigError) as raised: + self.module.resolve_auth( + SimpleNamespace(use_git_credential=False), self.target + ) + self.assertNotIn("token-secret", str(raised.exception)) + self.assertNotIn("injected-header", str(raised.exception)) + + def test_missing_git_credential_fails_without_echoing_helper_output(self): + credential = subprocess.CompletedProcess( + args=["git", "credential", "fill"], + returncode=1, + stdout="", + stderr="helper diagnostic containing basic-secret", + ) + with mock.patch.dict(os.environ, {}, clear=True): + with mock.patch.object( + self.module.subprocess, "run", return_value=credential + ): + with self.assertRaises(self.module.ConfigError) as raised: + self.module.resolve_auth( + SimpleNamespace(use_git_credential=True), self.target + ) + self.assertNotIn("basic-secret", str(raised.exception)) + + def test_git_credential_timeout_does_not_echo_helper_output(self): + timeout = subprocess.TimeoutExpired( + cmd=["git", "credential", "fill"], + timeout=10, + output="username=ci-user\npassword=basic-secret\n", + stderr="helper diagnostic containing basic-secret", + ) + with mock.patch.dict(os.environ, {}, clear=True): + with mock.patch.object(self.module.subprocess, "run", side_effect=timeout): + with self.assertRaises(self.module.ConfigError) as raised: + self.module.resolve_auth( + SimpleNamespace(use_git_credential=True), self.target + ) + self.assertNotIn("basic-secret", str(raised.exception)) + + def test_api_client_sends_auth_only_to_the_configured_origin(self): + opener = RecordingOpener() + auth = self.module.ApiAuth( + source="git credential", + authorization="Basic sensitive-value", + ) + client = self.module.ApiClient(self.target, auth=auth, opener=opener) + + status, body, _ = client.request( + self.target.repo_path("/actions/runs"), accept_json=True + ) + self.assertEqual(status, 200) + self.assertEqual(body, b'{"ok": true}') + request, timeout = opener.requests[0] + self.assertEqual(request.get_header("Authorization"), "Basic sensitive-value") + self.assertEqual(timeout, self.module.REQUEST_TIMEOUT) + + with self.assertRaisesRegex(self.module.ConfigError, "origin"): + client.request("https://evil.example/actions/runs", accept_json=True) + self.assertEqual(len(opener.requests), 1) + + def test_api_client_rejects_cross_origin_redirects(self): + auth = self.module.ApiAuth( + source="git credential", + authorization="Basic sensitive-value", + ) + handler = self.module.SameOriginRedirectHandler(self.target) + request = urllib.request.Request( + self.target.repo_path("/actions/runs"), + headers={"Authorization": auth.authorization}, + ) + + with self.assertRaises(self.module.ConfigError) as raised: + handler.redirect_request( + request, + None, + 302, + "Found", + {}, + "https://evil.example/actions/runs", + ) + self.assertNotIn("sensitive-value", str(raised.exception)) + + def test_401_diagnostics_distinguish_missing_and_rejected_auth(self): + anonymous = self.module._explain_status(401, self.target, None) + self.assertIn("GITEA_TOKEN", anonymous) + self.assertIn("--use-git-credential", anonymous) + + auth = self.module.ApiAuth( + source="git credential", + authorization="Basic sensitive-value", + ) + rejected = self.module._explain_status(401, self.target, auth) + self.assertIn("git credential", rejected) + self.assertNotIn("sensitive-value", rejected) + + def test_403_and_404_diagnostics_have_distinct_actions(self): + auth = self.module.ApiAuth( + source="GITEA_TOKEN", + authorization="token sensitive-value", + ) + forbidden = self.module._explain_status(403, self.target, auth) + self.assertIn("permission", forbidden) + self.assertIn("GITEA_TOKEN", forbidden) + + missing = self.module._explain_status(404, self.target, auth) + self.assertIn("endpoint", missing) + self.assertIn("version", missing) + self.assertNotIn("sensitive-value", forbidden + missing) + + def test_skill_documents_explicit_git_credential_auth(self): + skill = (SKILL_ROOT / "SKILL.md").read_text(encoding="utf-8") + self.assertIn("--use-git-credential", skill) + self.assertIn("GITEA_TOKEN", skill) + self.assertIn("HTTPS", skill) + + +if __name__ == "__main__": + unittest.main()