csh/playbook
1
1
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

✨ feat(gitea-fix-ci): add authenticated log collector

Browse Source
This commit is contained in:
csh
2026-07-22 17:28:22 +08:00
parent 38ea8caaeb
commit 3117ca5ae4
3 changed files with 1135 additions and 103 deletions
Download Patch File Download Diff File Expand all files Collapse all files
skills/gitea-fix-ci/SKILL.md
+162 -103
View File
@@ -1,151 +1,210 @@
--- ---
name: gitea-fix-ci 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 # Gitea Fix CI
## Overview ## 概述
Diagnose failing Gitea Actions from the pull request or workflow run, extract the 从 pull request 或 workflow run 诊断失败的 Gitea Actions，提取最小可用的失败上下文，
smallest useful failure context, then propose a fix plan before changing code. 然后在改代码之前先给出修复计划。核心原则：CI 日志是证据，不要仅凭红色状态臆测原因。
Core principle: CI logs are evidence; do not guess from the red status alone.
This is not a standalone executor. It guides use of `tea`, local `git`, and the 本 skill 不是独立执行器。它指导你在当前工作区中使用随附的取证脚本
Gitea API from the current workspace. `scripts/fetch_ci_logs.py`、`tea`、本地 `git` 和 Gitea API。
## When to Use 取证阶段（step 2-4）优先使用随附脚本，把「探测版本 → 找失败 run → 列 job →
下载失败 job 日志」这段易记错 API 路径的逻辑交给它。脚本仅取证到日志，分类、
修复计划与改代码仍由你按本文档执行。将 `<skill-dir>` 替换为包含本 `SKILL.md`
的目录：
- A Gitea-hosted repository has failing Gitea Actions or PR checks ```bash
- The user asks to inspect a failed workflow run, job, or CI status # 方式一（优先）：通过环境变量提供 API token
- The user asks to fix CI after a push, branch update, or pull request update export GITEA_TOKEN=<token>
- Local tests pass but remote Gitea Actions fail python <skill-dir>/scripts/fetch_ci_logs.py version # 探测实例版本并确认连通性
python <skill-dir>/scripts/fetch_ci_logs.py runs --status failure --branch <branch>
python <skill-dir>/scripts/fetch_ci_logs.py jobs <run-id> # 标注 failure 的 job
python <skill-dir>/scripts/fetch_ci_logs.py logs <job-id> # 日志存临时文件并回显尾部
## When Not to Use # 方式二：显式复用当前仓库的 Git HTTP 凭据
python <skill-dir>/scripts/fetch_ci_logs.py --use-git-credential version
python <skill-dir>/scripts/fetch_ci_logs.py --use-git-credential runs --status failure
```
- The repository is not hosted on Gitea or Forgejo-compatible infrastructure `base-url`、owner 和 repo 默认从 git remote 推导。认证规则如下：
- 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
## 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 脚本不可用（未装 Python、旧版 Gitea 无 job-log API、脱离 Gitea 环境）时，回退到
- Gitea base URL and repository owner/name, from `git remote -v` when possible 下文的手动 `tea`/API/web 路径。
- 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
## 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 login list`
- 使用前先运行 `tea actions --help` 确认已安装的 `tea` 版本存在 `actions`
子命令；并非所有版本都带这些子命令。
- `tea actions runs list --status failure --branch <branch>` - `tea actions runs list --status failure --branch <branch>`
- `tea actions runs view <run-id>` - `tea actions runs view <run-id>`
- `tea actions runs logs <run-id> --job <job-id>` - `tea actions runs logs <run-id> --job <job-id>`
- `tea pulls view <pr>` - `tea pulls view <pr>`
- If `tea` is unavailable or lacks an Actions command for the installed - 若 `tea` 不可用，或已安装版本缺少 Actions 命令，则直接使用 Gitea API。
version, use the Gitea API directly. - 当 API 行为不明确时，检查 `/api/v1/version` 和 `/swagger.v1.json`。较旧的
- Check `/api/v1/version` and `/swagger.v1.json` when API behavior is Gitea（1.21 及更早）在 web UI 中提供 Actions 页面，但不提供 workflow
unclear. Gitea 1.21 exposes Actions pages but not workflow run/job/log API run/job/log 的 API 端点；应对照上报的版本确认可用性，而不是想当然。
endpoints. - 绝不打印 token、用户名或密码。API token 只通过环境变量传递；Git 凭据只通过
- Never print tokens. Pass API tokens through environment variables. `git credential fill` 的标准输入/输出在脚本进程内传递，不放入 argv。
3. **Find failing workflow runs** 3. **定位失败的 workflow run**
- For a known run ID, fetch that run directly. - 若入口是 PR 而非 run ID，先把 PR 解析成 run：取 PR 的 head 分支与 head SHA
- Otherwise list recent workflow runs filtered by branch, event, status, or （`fetch_ci_logs.py` 的 `runs` 支持 `--branch`/`--sha`；手动则
commit SHA. `GET /api/v1/repos/<owner>/<repo>/pulls/<pr>` 取 `head.ref`/`head.sha`），
- API patterns: 再按该 SHA 过滤 runs。PR 页面的 checks 列表可能聚合多个 workflow，逐一定位到
具体失败 run，不要假设只有一个。
- 优先用脚本：`fetch_ci_logs.py runs --status failure --branch <branch>`
（也支持 `--sha`/`--event`/`--limit`），已知 run ID 时用
`fetch_ci_logs.py jobs <run-id>` 直接列出各 job 并标注失败项。
- 脚本不可用时回退到手动方式。已知 run ID 时，直接获取该 run。
- 否则按分支、事件、状态或 commit SHA 过滤，列出最近的 workflow runs。
- API 模式：
- `GET /api/v1/repos/<owner>/<repo>/actions/runs` - `GET /api/v1/repos/<owner>/<repo>/actions/runs`
- `GET /api/v1/repos/<owner>/<repo>/actions/runs/<run>` - `GET /api/v1/repos/<owner>/<repo>/actions/runs/<run>`
- `GET /api/v1/repos/<owner>/<repo>/actions/runs/<run>/jobs` - `GET /api/v1/repos/<owner>/<repo>/actions/runs/<run>/jobs`
- Web fallback for older Gitea: - 较旧 Gitea 的 web 回退：
- `GET /<owner>/<repo>/actions` - `GET /<owner>/<repo>/actions`
- Parse `/<owner>/<repo>/actions/runs/<run>` links and status labels. - 解析 `/<owner>/<repo>/actions/runs/<run>` 链接和状态标签。
- Treat `failure`, `cancelled`, and missing required checks differently. - 区别对待 `failure`、`cancelled` 和缺失的必需 checks。被取消（cancelled）的
Cancelled jobs may require rerun or queue investigation rather than code job 可能需要重跑或排查队列，而非改代码。
changes.
4. **Fetch job logs** 4. **获取 job 日志**
- For each failed job, download the job logs: - 优先用脚本：`fetch_ci_logs.py logs <job-id>`。它把日志存到受跟踪源码路径
之外的临时文件、只回显尾部若干行，并打印临时文件路径供进一步查看
（`--tail N` 调整行数，`--json` 输出结构化结果）。
- 脚本不可用时回退到手动方式。对每个失败的 job，下载其日志：
- `GET /api/v1/repos/<owner>/<repo>/actions/jobs/<job_id>/logs` - `GET /api/v1/repos/<owner>/<repo>/actions/jobs/<job_id>/logs`
- On Gitea 1.21 web fallback, download logs by UI job index: - 在较旧 Gitea 的 web 回退（无 job-log API）下，按 UI 的 job index 下载日志：
- `GET /<owner>/<repo>/actions/runs/<run>/jobs/<job-index>/logs` - `GET /<owner>/<repo>/actions/runs/<run>/jobs/<job-index>/logs`
- First open the run page and read each job's status; take the index of a - 先打开 run 页面读取每个 job 的状态；取状态为 `failure` 的 job 的 index，
job whose status is `failure`, not index `0` by default. The failing job 不要默认用 index `0`。失败的 job 很少是第一个，盲目用 index `0` 通常会返回
is rarely the first one, so a blind index `0` usually returns a passing 某个通过的 job 的日志。只有当 run 页面不暴露每个 job 的状态时，才回退到逐一
job's log. Only fall back to scanning indices when the run page does not 扫描 index。
expose per-job status. - 大日志保存到受跟踪源码路径之外的临时文件。
- Save large logs to a temporary file outside tracked source paths. - 提取首个可操作的错误块、其上下文命令、job 名、workflow 名、run URL、分支和 SHA。
- 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.
5. **Classify the failure** 5. **分类失败**
- Code/test failure: failing assertion, compile error, lint error, type error - 代码/测试失败：断言失败、编译错误、lint 错误、类型错误
- Environment failure: missing secret, runner image, dependency install, - 环境失败：缺失 secret、runner 镜像、依赖安装、网络、缓存、权限或服务启动
network, cache, permission, or service startup - 工作流失败：YAML 非法、语法不支持、触发条件错误、路径错误、分支/ref 假设错误
- Workflow failure: invalid YAML, unsupported syntax, wrong trigger, bad path, - 基础设施失败：runner 离线、队列卡住、run 被取消、超时
wrong branch/ref assumption
- Infrastructure failure: offline runner, stuck queue, cancelled run, timeout
6. **Create a fix plan** 6. **制定修复计划**
- Summarize the failure evidence with exact job/run identifiers. - 用确切的 job/run 标识总结失败证据。
- 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. - 在用户批准修复计划之前不要实施。
7. **Implement after approval** 7. **批准后实施**
- Apply only the approved fix. - 只应用已批准的修复。
- Run the local command that most closely reproduces the failed job. - 运行最接近复现该失败 job 的本地命令。
- If the failure is workflow-only, validate the workflow file syntax and any - 若失败仅涉及工作流，验证工作流文件语法及其引用的路径或脚本。
referenced paths or scripts.
8. **Recheck** 8. **复检**
- Tell the user what must be pushed or rerun in Gitea. - 告诉用户需要在 Gitea 中 push 或重跑什么。
- If permitted, use the Gitea API to inspect the rerun status. - 若获准，使用 Gitea API 检查重跑状态。
- Final output must distinguish local verification from remote CI status. - 最终输出必须区分本地验证与远端 CI 状态。
## Output Contract ## 输出约定
- `Target:` repo, branch/SHA, PR or run ID - `Target:` 仓库、分支/SHA、PR 或 run ID
- `Failed CI:` workflow, job, status, run URL or API path - `Failed CI:` workflow、job、状态、run URL 或 API 路径
- `Evidence:` concise log snippet and classification - `Evidence:` 精简的日志片段与分类
- `Plan:` proposed fix, local verification, remote recheck - `Plan:` 提出的修复、本地验证、远端复检
- `Changes:` files changed after approval - `Changes:` 批准后改动的文件
- `Result:` local checks run and remaining remote status - `Result:` 已运行的本地检查与剩余的远端状态
## Success Criteria ## 成功标准
- Failure analysis is based on Gitea Actions run/job data or pasted logs - 失败分析基于 Gitea Actions 的 run/job 数据或粘贴的日志
- The fix plan names the exact workflow run or job it addresses - 修复计划指明其针对的确切 workflow run 或 job
- No code or workflow edits happen before plan approval - 计划批准前不发生任何代码或工作流改动
- Verification distinguishes local commands from remote Gitea Actions results - 验证区分本地命令与远端 Gitea Actions 结果
- Tokens and private log content are not echoed unnecessarily - 不回显 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 - **凭红色状态臆测原因**：还没下载 job 日志就断言失败原因或动手改代码。
or a browser-copied job log - **误读双字段结果**：把 `status`（生命周期）当成结果判定。Gitea Actions 沿用
- If an external CI provider owns the failing check, report the external URL and GitHub 兼容的双字段模型——`status=completed` 只表示跑完了，真正的成败在
stop at evidence collection `conclusion`（`failure`/`success`/`cancelled`）。判定失败必须看有效结果，
- If the failure is infrastructure-only, recommend rerun/runner investigation 而非 `status=completed` 就当通过。
instead of editing code - **盲取 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，而非改代码
skills/gitea-fix-ci/scripts/fetch_ci_logs.py
+699
View File
@@ -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"^(?P<scheme>https?)://(?:[^@/]+@)?(?P<host>[^/]+)/"
r"(?P<owner>[^/]+)/(?P<repo>[^/]+?)(?:\.git)?/?$"
)
_SCP_REMOTE_RE = re.compile(
r"^(?:ssh://)?(?:[^@]+@)?(?P<host>[^:/]+)(?::\d+)?[:/]"
r"(?P<owner>[^/]+)/(?P<repo>[^/]+?)(?:\.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 <host>/<owner>/<repo>; 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, "<redacted>")
if secret:
detail = detail.replace(secret, "<redacted>")
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 <job-id>")
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())
test/test_gitea_fix_ci_skill.py
+274
View File
@@ -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()