release: prepare version 1.0.0
This commit is contained in:
cmx278171032 committed
1 parent
98262e83e8
commit
468b13e105
10 files changed
+1
-671
No files matched your search
@@ -1,6 +1,6 @@
|
||||
{
|
||||
"name": "updating-gitea-repositories",
|
||||
"version": "0.2.0",
|
||||
"version": "1.0.0",
|
||||
"description": "通过安全凭据持久复用、Git Credential Manager 和 HTTPS 更新 Gitea 仓库",
|
||||
"author": {
|
||||
"name": "Eapil"
|
||||
|
||||
@@ -1,61 +0,0 @@
|
||||
# Gitea 凭据持久化实施计划
|
||||
|
||||
> **面向 Agent 型工作者：** 按任务验证后再进行真实环境操作；用户已确认设计和实施范围。
|
||||
|
||||
**目标：** 让同一执行用户的 Gitea 凭据安全持久化、跨仓库复用，并减少失效时的重复授权。
|
||||
|
||||
**架构：** 保留现有 skill，以 GCM 和平台安全凭据库为基础；将可验证的凭据迁移、诊断和 Windows OAuth 存储隔离到小工具。按实际有效配置诊断认证问题，OAuth 仅在兼容性经过验证时启用。
|
||||
|
||||
**技术栈：** Git、GCM、Python 标准库、各平台凭据库和 Markdown。
|
||||
|
||||
**规格说明：** `docs/superpowers/specs/2026-11-05-gitea-credential-persistence-design.md`
|
||||
|
||||
## 全局约束
|
||||
|
||||
- 用户和站点级复用；保留其他站点配置与旧凭据。
|
||||
- 密钥不得出现在聊天、日志、命令参数、URL、环境变量或项目文件。
|
||||
- 实际认证测试不能意外删除或覆盖旧凭据。
|
||||
- 示例仓库不执行发布或推送；skill 项目验证通过后推送。
|
||||
- 站点不支持 OAuth 或无法证明兼容时，保留可用的 PAT 路径。
|
||||
|
||||
## 审查重点
|
||||
|
||||
1. Git 配置包含不同优先级和多个 helper 时，真实生效配置不能被误判。
|
||||
2. 源账号或站点不匹配、多条凭据存在歧义时，迁移须停止。
|
||||
3. 目标凭据已存在且不相同时，不能覆盖；源凭据始终保留。
|
||||
4. 失败或拒绝的认证请求不得泄露或删除凭据。
|
||||
5. 无图形会话或系统库锁定时，不能把短期缓存冒充持久化。
|
||||
|
||||
## Task 1: 验证并实现安全迁移
|
||||
|
||||
**文件：** `skills/updating-gitea-repositories/scripts/credential_migrate.py`；`tests/test_credential_migrate.py`。
|
||||
|
||||
- [x] 在隔离配置中用虚构 helper 和虚构凭据，验证 source `get`、target `store` 和跨仓库凭据匹配。
|
||||
- [x] 验证目标已存在、缺少密码、账号或站点不匹配时停止；输出不含虚构密钥；源 helper 不收到 `erase` 或 `store`。迁移总是选用新的隔离命名空间，不覆盖已有目标。
|
||||
- [x] 实现只接收明确源文件和 HTTPS 站点的迁移工具；通过受限子进程和内存通道传递密钥，失败时只输出非敏感原因。
|
||||
- [x] 用真实 Git 和隔离的虚构凭据运行测试，确认目标写入、原记录保留和失败保护。
|
||||
|
||||
## Task 2: 修订 skill 和平台指南
|
||||
|
||||
**文件：** `skills/updating-gitea-repositories/SKILL.md`；`references/repository-workflow.md`、`windows.md`、`macos.md`、`linux.md`；必要时新增 `references/credentials.md`；`.codex-plugin/plugin.json`。
|
||||
|
||||
- [x] 改为先检查和复用已有凭据，再按站点修复配置；补充不会误删凭据的诊断、迁移和 401 分流。
|
||||
- [x] Windows、macOS、WSL、Linux 指明安全持久化与实际运行位置；无图形 Linux 采用可用的持久库，说明初始化与解锁。
|
||||
- [x] 纳入条件式 OAuth 流程和 2.6.1 的刷新风险；明确需验证站点、客户端版本、权限和客户端注册，禁止写出客户端密钥。
|
||||
- [x] 更新插件版本与展示说明，并检查各入口、参考链接和脚本调用一致。
|
||||
|
||||
## Task 3: 验证示例和交付
|
||||
|
||||
**文件：** 上述 skill 和测试；ManySkills 只允许已授权的认证配置处理。
|
||||
|
||||
- [x] 运行安全迁移测试、skill 结构验证；对认证决策做独立情景检查。
|
||||
- [x] 不暴露凭据、不触发旧凭据删除地检查 ManySkills；通过浏览器完成一次授权并验证跨仓库复用。
|
||||
- [ ] 确认目标项目的暂存范围、测试和远端状态，正常提交并推送，获取远端 SHA 核对结果。
|
||||
- [ ] 核对市场来源锁定的上游提交及安装缓存，分别报告项目推送和市场同步状态；未经授权不修改其他市场来源。
|
||||
|
||||
## 补充：OIDC 用户与长 OAuth 令牌
|
||||
|
||||
- [x] 验证真实 Gitea 站点的浏览器 OAuth、OIDC 账号及 Git 读取权限。
|
||||
- [x] 处理 Windows 自带 helper 无法同时保存长访问和刷新令牌的问题，安全复用系统凭据库并保留原记录。
|
||||
- [x] 使用隔离虚构服务验证过期后的 Git 自动刷新、轮换结果持久化及自动 erase 不删除凭据。
|
||||
- [x] 验证真实站点刷新后，ManySkills 与 skill 项目可读取新凭据；不推送 ManySkills。
|
||||
@@ -1,62 +0,0 @@
|
||||
# Gitea 凭据持久化设计
|
||||
|
||||
## 目标
|
||||
|
||||
让 AI 使用 `updating-gitea-repositories` 时，优先复用执行 Git 的机器上已有的安全凭据。对于同一系统用户、Gitea 站点和账号，同一个有效凭据应能用于该账号有权访问的多个仓库。首次配置、令牌失效或安全库被锁定时，仍可能需要用户参与。
|
||||
|
||||
本次同时处理 ManySkills 示例环境中的认证问题；ManySkills 的发布与推送继续遵照原会话的安排。完成验证后，将 skill 更新推送到本项目仓库，并核实市场获取更新的方式。
|
||||
|
||||
## 方案
|
||||
|
||||
用户选择扩展现有 skill，复用 Git Credential Manager（GCM）和操作系统提供的安全凭据库。仅在凭据迁移、安全诊断或平台兼容需要确定性处理时，增加范围明确、可验证的小型辅助脚本。
|
||||
|
||||
备选方案是开发独立凭据管理器。该方案能够统一控制凭据生命周期，但会引入额外的敏感代码、跨平台兼容和维护成本，本次不采用。Windows 的 OAuth 存储适配器只调用系统凭据库，不实现独立密码库或 OAuth 客户端。
|
||||
|
||||
## 凭据复用和配置
|
||||
|
||||
- 执行 Git 前，检查实际生效的站点、账号、helper、凭据库及路径匹配配置；考虑系统、用户、仓库和站点级设置的覆盖关系。
|
||||
- 优先使用已有的有效凭据。只有确认缺少可用凭据或需要新的授权时，才引导用户进行相应操作；不无条件要求创建新 PAT。
|
||||
- 允许 AI 自动修复该 Gitea 站点所需的当前用户级配置，并保留其他站点的设置和已有凭据。账号或凭据匹配存在歧义时，先澄清，避免覆盖或使用错误账号。
|
||||
- 同一站点的多个仓库默认复用凭据，但仍尊重仓库访问权限、站点策略和用户明确配置的账号隔离要求。
|
||||
- 已有旧凭据与目标账号匹配明确时，可以通过不展示凭据内容的方式复制到可用的安全凭据库；不得将密钥输出到聊天、日志、命令参数、URL、环境变量或项目文件。迁移前后保留原记录。如果原记录为明文，应告知明文副本仍然存在。
|
||||
- 诊断和迁移不得未经授权删除或覆盖原凭据。考虑 Git 在认证失败时可能向 helper 发送删除请求，验证方案须防止意外清除现有凭据；无法保证时停止并说明原因。
|
||||
|
||||
## 不同环境
|
||||
|
||||
- Windows 优先使用 GCM 和 Windows Credential Manager；macOS 优先使用 GCM 和 Keychain。
|
||||
- WSL 或远程环境依据 Git 实际运行的主机及可用凭据库选择方案，不根据控制端操作系统推断。
|
||||
- Linux 优先使用现有可用的安全凭据库；无图形会话需要持久化时，可安装必要组件并引导初始化 GPG/pass 或其他适用的安全存储。首次设置和后续解锁可能需要用户参与。
|
||||
- 不把 15 分钟内存缓存当作长期持久化方案，也不静默退回明文存储。
|
||||
|
||||
## 认证方式和故障处理
|
||||
|
||||
- 保留 GCM + PAT。站点和客户端确实支持时，可使用浏览器 OAuth 登录及令牌刷新；具体流程、权限范围、客户端注册和跨版本兼容性必须先验证。不得为了配置 OAuth 将客户端密钥或令牌明文写入 Git 配置。
|
||||
- 仅在确认当前 GCM 版本阻碍所需功能时，按需更新到与系统兼容的稳定版本；普通仓库更新继续复用现有安装。
|
||||
- 对认证失败，先区分凭据匹配、客户端配置、网络、服务端认证和仓库权限等可观察问题。HTTP 401 本身不足以证明令牌已过期；无法确定时明确说明，不盲目创建或撤销令牌。
|
||||
- 令牌过期、被撤销或用户失去权限时，持久化不能代替重新授权。需要用户输入时，通过本地可信交互完成，不要求用户在聊天中发送密钥。
|
||||
|
||||
## 示例环境的已知事实
|
||||
|
||||
ManySkills 示例此前的凭据被 Gitea 拒绝，返回 HTTP 401。当前站点级配置使用 GCM、Windows 凭据管理器和 `useHttpPath=false`；该值覆盖仓库级通用的 `true`。现有证据不能确定 401 的具体原因。
|
||||
|
||||
本机 GCM 为 2.6.1。GCM 2.7.0 修复了通用 OAuth 的刷新令牌更新问题；这不是当前 PAT 认证失败的已证实原因。OAuth 路径是否受影响及是否需要升级须单独验证。
|
||||
|
||||
用户确认常用 OIDC 登录；该站点支持公开 OAuth 客户端和 PKCE。浏览器授权后，用户身份及两个仓库的认证读取得到验证。Git 自带 Windows 凭据助手将访问与刷新令牌合并为 UTF-16 后超过系统凭据条目大小限制；增加隔离的 Windows 凭据库适配器，以单个压缩条目保存令牌和到期时间，不使用明文文件。
|
||||
|
||||
以隔离虚构服务验证 Git 在令牌过期时自动刷新和保存轮换结果，并对真实 Gitea 执行一次刷新和安全库回读；真实到期时的自动刷新仍取决于 Git 和服务器的后续行为。原有凭据被保留，ManySkills 未推送。
|
||||
|
||||
## 验收
|
||||
|
||||
1. 对已有有效凭据的场景，AI 能识别并复用，避免重复要求用户创建或输入 PAT。
|
||||
2. 使用隔离的 Git 配置及虚构凭据验证站点级覆盖、跨仓库匹配、不同账号、迁移保留原记录，以及输出和日志不泄露密钥。失败的认证探测不得删除原凭据。
|
||||
3. 覆盖无图形 Linux 的持久化决策、凭据库锁定，以及缺少组件时的初始化流程；不能将短期缓存冒充持久化。
|
||||
4. 区分 401、仓库权限不足和网络故障；无法确定原因时不宣称具体根因。验证实际支持的 OAuth 流程；不支持时保留 PAT 路径。
|
||||
5. 验证 skill 的元数据、引用和关键行为。完成已授权的 ManySkills 认证处理后，使用不推送该仓库的方式确认可用；如必须由用户重新授权，明确给出安全操作步骤。
|
||||
6. 推送本 skill 项目前确认变更范围并验证目标远端提交。分别报告项目仓库推送结果和市场同步结果，不假定两者等同。
|
||||
|
||||
## 参考
|
||||
|
||||
- [GCM 凭据库](https://github.com/git-ecosystem/git-credential-manager/blob/main/docs/credstores.md)
|
||||
- [GCM 通用 OAuth](https://github.com/git-ecosystem/git-credential-manager/blob/main/docs/generic-oauth.md)
|
||||
- [GCM 2.7.0 发布说明](https://github.com/git-ecosystem/git-credential-manager/releases/tag/v2.7.0)
|
||||
- [Git 凭据配置](https://git-scm.com/docs/gitcredentials)
|
||||
@@ -1,171 +0,0 @@
|
||||
import contextlib
|
||||
from http.server import BaseHTTPRequestHandler, HTTPServer
|
||||
import threading
|
||||
import importlib.util
|
||||
import io
|
||||
import os
|
||||
from pathlib import Path
|
||||
import shutil
|
||||
import subprocess
|
||||
import tempfile
|
||||
import unittest
|
||||
|
||||
|
||||
SCRIPT = Path(__file__).resolve().parents[1] / "skills" / "updating-gitea-repositories" / "scripts" / "credential_migrate.py"
|
||||
spec = importlib.util.spec_from_file_location("credential_migrate", SCRIPT)
|
||||
credential_migrate = importlib.util.module_from_spec(spec)
|
||||
spec.loader.exec_module(credential_migrate)
|
||||
|
||||
SITE = "https://gitea.invalid"
|
||||
ACCOUNT = "fictional-alice"
|
||||
SECRET = b"fictional-test-secret-do-not-use"
|
||||
|
||||
|
||||
class FakeHelper:
|
||||
def __init__(self, response=None, after_store=None):
|
||||
self.response = response
|
||||
self.after_store = after_store
|
||||
self.operations = []
|
||||
self.stored = None
|
||||
|
||||
def get(self, request):
|
||||
self.operations.append("get")
|
||||
if self.stored is not None:
|
||||
return self.after_store if self.after_store is not None else self.stored
|
||||
return self.response
|
||||
|
||||
def store(self, credential):
|
||||
self.operations.append("store")
|
||||
self.stored = dict(credential)
|
||||
|
||||
|
||||
class MigrationTests(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.good = {"username": ACCOUNT.encode(), "password": SECRET}
|
||||
self.source = FakeHelper(self.good)
|
||||
self.target = FakeHelper()
|
||||
|
||||
def migrate(self, site=SITE, account=ACCOUNT):
|
||||
return credential_migrate.migrate(site, account, self.source, self.target)
|
||||
|
||||
def test_copies_without_touching_source_and_verifies_target(self):
|
||||
self.migrate()
|
||||
self.assertEqual(self.source.operations, ["get"])
|
||||
self.assertEqual(self.target.operations, ["get", "store", "get"])
|
||||
self.assertTrue(self.target.stored["password"] == SECRET, "stored credential differs")
|
||||
|
||||
def test_existing_target_is_not_overwritten(self):
|
||||
self.target.response = {"username": b"other", "password": b"other-fake-secret"}
|
||||
with self.assertRaises(credential_migrate.MigrationError):
|
||||
self.migrate()
|
||||
self.assertEqual(self.target.operations, ["get"])
|
||||
|
||||
def test_refuses_missing_or_wrong_source_account_and_password(self):
|
||||
for response in (
|
||||
{"password": SECRET},
|
||||
{"username": b"different-account", "password": SECRET},
|
||||
{"username": ACCOUNT.encode()},
|
||||
{"username": ACCOUNT.encode(), "password": b""},
|
||||
{"username": ACCOUNT.encode(), "password": SECRET, "host": b"other.invalid"},
|
||||
{"username": ACCOUNT.encode(), "password": SECRET, "protocol": b"http"},
|
||||
):
|
||||
with self.subTest(fields=tuple(response.keys())):
|
||||
self.source = FakeHelper(response)
|
||||
self.target = FakeHelper()
|
||||
with self.assertRaises(credential_migrate.MigrationError):
|
||||
self.migrate()
|
||||
self.assertNotIn("store", self.target.operations)
|
||||
|
||||
def test_refuses_unsafe_url_and_account_before_accessing_helpers(self):
|
||||
for site, account in (
|
||||
("http://gitea.invalid", ACCOUNT),
|
||||
("https://name:secret@gitea.invalid", ACCOUNT),
|
||||
("https://gitea.invalid/repo", ACCOUNT),
|
||||
("https://gitea.invalid?secret=1", ACCOUNT),
|
||||
(SITE, "alice\npassword=leak"),
|
||||
):
|
||||
with self.subTest(site=site.split("@")[0], account=account.splitlines()[0]):
|
||||
self.source = FakeHelper(self.good)
|
||||
self.target = FakeHelper()
|
||||
with self.assertRaises(credential_migrate.MigrationError):
|
||||
self.migrate(site, account)
|
||||
self.assertEqual(self.source.operations, [])
|
||||
self.assertEqual(self.target.operations, [])
|
||||
|
||||
def test_failed_readback_is_not_reported_as_success(self):
|
||||
self.target.after_store = {"username": ACCOUNT.encode(), "password": b"wrong-fake-secret"}
|
||||
with self.assertRaises(credential_migrate.MigrationError):
|
||||
self.migrate()
|
||||
self.assertEqual(self.source.operations, ["get"])
|
||||
self.assertNotIn("erase", self.target.operations)
|
||||
|
||||
def test_helper_failure_message_does_not_leak_secret(self):
|
||||
class BadHelper(FakeHelper):
|
||||
def get(self, request):
|
||||
raise RuntimeError(SECRET.decode())
|
||||
self.source = BadHelper()
|
||||
with self.assertRaises(credential_migrate.MigrationError) as caught:
|
||||
self.migrate()
|
||||
self.assertNotIn(SECRET.decode(), str(caught.exception))
|
||||
|
||||
def test_explicit_source_rejects_path_scoped_and_ambiguous_records(self):
|
||||
with tempfile.TemporaryDirectory() as directory:
|
||||
source_file = Path(directory) / "credentials"
|
||||
for contents in (
|
||||
"https://fictional-alice:fictional-secret@gitea.invalid/team/first.git\n",
|
||||
"https://fictional-alice:fictional-secret@gitea.invalid\nhttps://fictional-alice:another-secret@gitea.invalid\n",
|
||||
):
|
||||
source_file.write_text(contents, encoding="utf-8")
|
||||
with self.assertRaises(credential_migrate.MigrationError) as caught:
|
||||
credential_migrate.inspect_source_file(SITE, ACCOUNT, source_file)
|
||||
self.assertNotIn("fictional-secret", str(caught.exception))
|
||||
source_file.write_text("https://fictional-alice:fictional-secret@gitea.invalid\n", encoding="utf-8")
|
||||
credential_migrate.inspect_source_file(SITE, ACCOUNT, source_file)
|
||||
def test_git_http_does_not_emit_curl_trace(self):
|
||||
git = shutil.which("git")
|
||||
if not git:
|
||||
self.skipTest("Git is not available")
|
||||
class Handler(BaseHTTPRequestHandler):
|
||||
def do_GET(self):
|
||||
self.send_response(404)
|
||||
self.end_headers()
|
||||
def log_message(self, *args):
|
||||
pass
|
||||
server = HTTPServer(("127.0.0.1", 0), Handler)
|
||||
worker = threading.Thread(target=server.serve_forever, daemon=True)
|
||||
worker.start()
|
||||
try:
|
||||
with tempfile.TemporaryDirectory() as directory:
|
||||
env = credential_migrate.safe_environment()
|
||||
env.update({"GIT_CONFIG_NOSYSTEM": "1", "GIT_CONFIG_GLOBAL": os.devnull})
|
||||
proc = subprocess.run([git, "-c", "credential.helper=", "ls-remote", "http://127.0.0.1:" + str(server.server_port) + "/repo.git"], cwd=directory, env=env, stdout=subprocess.PIPE, stderr=subprocess.PIPE, timeout=10, check=False)
|
||||
self.assertNotIn(b"=> Send header", proc.stderr, "HTTP tracing was enabled")
|
||||
finally:
|
||||
server.shutdown()
|
||||
server.server_close()
|
||||
worker.join()
|
||||
def test_git_store_source_keeps_old_record_and_reuses_across_paths(self):
|
||||
git = shutil.which("git")
|
||||
if not git:
|
||||
self.skipTest("Git is not available")
|
||||
with tempfile.TemporaryDirectory() as directory:
|
||||
source_file = Path(directory) / "source"
|
||||
target_file = Path(directory) / "target"
|
||||
base_env = credential_migrate.safe_environment()
|
||||
source = credential_migrate.GitHelper([git, "credential-store", "--file=" + str(source_file)], base_env, directory)
|
||||
target = credential_migrate.GitHelper([git, "credential-store", "--file=" + str(target_file)], base_env, directory)
|
||||
original = {"protocol": b"https", "host": b"gitea.invalid", "username": ACCOUNT.encode(), "password": SECRET}
|
||||
source.store(original)
|
||||
before = source_file.read_bytes()
|
||||
credential_migrate.migrate(SITE, ACCOUNT, source, target)
|
||||
self.assertEqual(source_file.read_bytes(), before)
|
||||
for path in ("team/first.git", "team/second.git"):
|
||||
config = "!" + "git credential-store --file='" + str(target_file).replace("\\", "/") + "'"
|
||||
request = ("url=" + SITE + "/" + path + "\n\n").encode()
|
||||
completed = subprocess.run([git, "-c", "credential.helper=", "-c", "credential.helper=" + config, "credential", "fill"], input=request, stdout=subprocess.PIPE, stderr=subprocess.PIPE, env=base_env, cwd=directory, check=False)
|
||||
self.assertEqual(completed.returncode, 0, "isolated git lookup failed")
|
||||
self.assertTrue(b"password=" + SECRET + b"\n" in completed.stdout, "credential not reused")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -1,55 +0,0 @@
|
||||
"""A browser OAuth credential is stored only after account and Git checks."""
|
||||
import importlib.util
|
||||
from pathlib import Path
|
||||
import sys
|
||||
import unittest
|
||||
from unittest.mock import patch
|
||||
|
||||
|
||||
SCRIPTS = Path(__file__).resolve().parents[1] / "skills" / "updating-gitea-repositories" / "scripts"
|
||||
sys.path.insert(0, str(SCRIPTS))
|
||||
spec = importlib.util.spec_from_file_location("credential_oauth_bootstrap_test", SCRIPTS / "credential_oauth_bootstrap.py")
|
||||
bootstrap = importlib.util.module_from_spec(spec)
|
||||
spec.loader.exec_module(bootstrap)
|
||||
|
||||
|
||||
class BootstrapTest(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.fields = {"username": "oauth2", "password": "fake-access", "password_expiry_utc": "9999999999", "oauth_refresh_token": "fake-refresh"}
|
||||
self.namespace = "gitea-oauth-" + "a" * 32
|
||||
self.store = {}
|
||||
|
||||
def run_bootstrap(self, api="matched_account", git="read_advertisement", occupied=False):
|
||||
reads = ([{"existing": "entry"}] if occupied else [None, self.fields])
|
||||
with patch.object(bootstrap.probe, "check_requests", side_effect=[(api, "ignored"), ("ignored", git)]), \
|
||||
patch.object(bootstrap.vault, "read_record", side_effect=reads), \
|
||||
patch.object(bootstrap.vault, "write_record") as write:
|
||||
try:
|
||||
bootstrap.persist_verified("https://example.invalid", "alice", "https://example.invalid/team/repo.git", self.fields, self.namespace, object())
|
||||
finally:
|
||||
self.store["write"] = write.call_count
|
||||
|
||||
def test_saves_after_matching_identity_and_git_read(self):
|
||||
self.run_bootstrap()
|
||||
self.assertEqual(self.store["write"], 1)
|
||||
|
||||
def test_rejects_wrong_identity_or_denied_git_without_storing(self):
|
||||
for api, git in [("identity_mismatch_or_unexpected_response", "read_advertisement"), ("matched_account", "http_401")]:
|
||||
with self.subTest(api=api, git=git), self.assertRaises(bootstrap.BootstrapError):
|
||||
self.run_bootstrap(api, git)
|
||||
self.assertEqual(self.store["write"], 0)
|
||||
|
||||
def test_refuses_to_replace_an_existing_vault_entry(self):
|
||||
with self.assertRaises(bootstrap.BootstrapError):
|
||||
self.run_bootstrap(occupied=True)
|
||||
self.assertEqual(self.store["write"], 0)
|
||||
|
||||
def test_rejects_incomplete_refreshable_credential(self):
|
||||
del self.fields["oauth_refresh_token"]
|
||||
with self.assertRaises(bootstrap.BootstrapError):
|
||||
self.run_bootstrap()
|
||||
self.assertEqual(self.store["write"], 0)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -1,35 +0,0 @@
|
||||
import importlib.util
|
||||
import io
|
||||
from pathlib import Path
|
||||
import sys
|
||||
import unittest
|
||||
from unittest.mock import patch
|
||||
|
||||
|
||||
SCRIPTS = Path(__file__).resolve().parents[1] / "skills" / "updating-gitea-repositories" / "scripts"
|
||||
sys.path.insert(0, str(SCRIPTS))
|
||||
spec = importlib.util.spec_from_file_location("credential_oauth_identity_test", SCRIPTS / "credential_oauth_identity.py")
|
||||
identity = importlib.util.module_from_spec(spec)
|
||||
spec.loader.exec_module(identity)
|
||||
|
||||
|
||||
class IdentityTest(unittest.TestCase):
|
||||
def test_reports_api_login_without_disclosing_token(self):
|
||||
class Response:
|
||||
def __enter__(self): return self
|
||||
def __exit__(self, *args): pass
|
||||
def read(self, limit): return b'{"login":"alice"}'
|
||||
class Opener:
|
||||
def open(self, request, timeout): return Response()
|
||||
def get(namespace, operation, fields, output):
|
||||
output.write(b"username=oauth2\npassword=fake-secret-token\n")
|
||||
with patch.object(identity.vault, "run", side_effect=get):
|
||||
self.assertEqual(identity.lookup("https://example.invalid", "gitea-oauth-" + "a" * 32, Opener()), "alice")
|
||||
|
||||
def test_empty_vault_result_is_rejected(self):
|
||||
with patch.object(identity.vault, "run"), self.assertRaises(identity.IdentityError):
|
||||
identity.lookup("https://example.invalid", "gitea-oauth-" + "a" * 32, object())
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -1,41 +0,0 @@
|
||||
import importlib.util
|
||||
from pathlib import Path
|
||||
import unittest
|
||||
from unittest.mock import patch
|
||||
from urllib.error import HTTPError
|
||||
|
||||
SCRIPT = Path(__file__).resolve().parents[1] / "skills" / "updating-gitea-repositories" / "scripts" / "credential_probe.py"
|
||||
spec = importlib.util.spec_from_file_location("credential_probe", SCRIPT)
|
||||
credential_probe = importlib.util.module_from_spec(spec)
|
||||
spec.loader.exec_module(credential_probe)
|
||||
|
||||
|
||||
class ProbeTests(unittest.TestCase):
|
||||
def test_rejects_different_or_credential_bearing_remote(self):
|
||||
for remote in ("https://other.invalid/a/b.git", "https://user:fake-secret@gitea.invalid/a/b.git", "https://gitea.invalid/a/b.git?token=fake-secret"):
|
||||
with self.subTest(case=remote.split("@")[0]):
|
||||
with self.assertRaises(credential_probe.ProbeError):
|
||||
credential_probe.git_read_url("https://gitea.invalid", remote)
|
||||
|
||||
def test_redirect_is_not_followed(self):
|
||||
handler = credential_probe.NoRedirect()
|
||||
self.assertIsNone(handler.redirect_request(None, None, 302, "", {}, "https://other.invalid"))
|
||||
|
||||
def test_unauthorized_is_reported_without_credentials(self):
|
||||
class RejectingOpener:
|
||||
def open(self, request, timeout):
|
||||
raise HTTPError(request.full_url, 401, "unauthorized", {}, None)
|
||||
result = credential_probe.check_requests("https://gitea.invalid", "fictional-alice", b"fictional-secret", "https://gitea.invalid/team/repo.git", RejectingOpener())
|
||||
self.assertEqual(result, ("http_401", "http_401"))
|
||||
self.assertNotIn("fictional-secret", str(result))
|
||||
|
||||
def test_different_account_is_rejected_before_network(self):
|
||||
class Helper:
|
||||
def get(self, request):
|
||||
return {"username": b"different", "password": b"fictional-secret"}
|
||||
with self.assertRaises(credential_probe.ProbeError):
|
||||
credential_probe.load_credential("https://gitea.invalid", "fictional-alice", Helper())
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -1,42 +0,0 @@
|
||||
import contextlib
|
||||
import importlib.util
|
||||
import io
|
||||
from pathlib import Path
|
||||
import unittest
|
||||
from unittest.mock import patch
|
||||
|
||||
|
||||
SCRIPT = Path(__file__).resolve().parents[1] / "skills" / "updating-gitea-repositories" / "scripts" / "credential_win_oauth.py"
|
||||
spec = importlib.util.spec_from_file_location("credential_win_oauth_unit_test", SCRIPT)
|
||||
vault = importlib.util.module_from_spec(spec)
|
||||
spec.loader.exec_module(vault)
|
||||
|
||||
|
||||
class VaultHelperTest(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.namespace = "gitea-oauth-" + "a" * 32
|
||||
self.fields = {"protocol": "https", "host": "example.invalid", "username": "oauth2",
|
||||
"password": "fake-access", "password_expiry_utc": "9999999999", "oauth_refresh_token": "fake-refresh"}
|
||||
|
||||
def test_empty_namespace_cannot_get_or_store_unverified_identity(self):
|
||||
with patch.object(vault, "credential_lock", return_value=contextlib.nullcontext()), \
|
||||
patch.object(vault, "read_record", return_value=None), \
|
||||
patch.object(vault, "write_record") as write:
|
||||
for operation in ("get", "store"):
|
||||
with self.subTest(operation=operation), self.assertRaises(vault.VaultError):
|
||||
vault.run(self.namespace, operation, self.fields, io.BytesIO())
|
||||
write.assert_not_called()
|
||||
|
||||
def test_refuses_cross_site_refresh_endpoint_before_sending_token(self):
|
||||
with patch.object(vault, "oauth_setting", side_effect=["client", "https://other.invalid/token"]), \
|
||||
patch.object(vault, "build_opener") as opener, self.assertRaises(vault.VaultError):
|
||||
vault.refresh_record("https", "example.invalid", self.fields)
|
||||
opener.assert_not_called()
|
||||
|
||||
def test_missing_context_fails_closed(self):
|
||||
with self.assertRaises(vault.VaultError):
|
||||
vault.run(self.namespace, "get", {"protocol": "https", "host": "example.invalid"}, io.BytesIO())
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -1,155 +0,0 @@
|
||||
"""Exercise OAuth refresh using only a fake server and disposable vault entry."""
|
||||
import base64
|
||||
from concurrent.futures import ThreadPoolExecutor
|
||||
import ctypes
|
||||
from ctypes import wintypes
|
||||
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
||||
import importlib.util
|
||||
import json
|
||||
import os
|
||||
from pathlib import Path
|
||||
import secrets
|
||||
import shutil
|
||||
import subprocess
|
||||
import sys
|
||||
import tempfile
|
||||
import threading
|
||||
import time
|
||||
import unittest
|
||||
from urllib.parse import parse_qs
|
||||
|
||||
|
||||
SCRIPT = Path(__file__).resolve().parents[1] / "skills" / "updating-gitea-repositories" / "scripts" / "credential_win_oauth.py"
|
||||
sys.path.insert(0, str(SCRIPT.parent))
|
||||
|
||||
|
||||
@unittest.skipUnless(sys.platform == "win32" and os.environ.get("GITEA_TEST_WINDOWS_OAUTH") == "1", "opt-in Windows OAuth test")
|
||||
class WindowsOAuthRefreshTest(unittest.TestCase):
|
||||
def test_expired_long_token_is_refreshed_and_saved(self):
|
||||
spec = importlib.util.spec_from_file_location("credential_win_oauth_test", SCRIPT)
|
||||
vault = importlib.util.module_from_spec(spec)
|
||||
spec.loader.exec_module(vault)
|
||||
bootstrap_spec = importlib.util.spec_from_file_location("oauth_bootstrap_for_refresh_test", SCRIPT.parent / "credential_oauth_bootstrap.py")
|
||||
bootstrap = importlib.util.module_from_spec(bootstrap_spec)
|
||||
bootstrap_spec.loader.exec_module(bootstrap)
|
||||
namespace = "gitea-oauth-" + secrets.token_hex(16)
|
||||
initial = base64.urlsafe_b64encode(secrets.token_bytes(642)).rstrip(b"=")
|
||||
fresh = base64.urlsafe_b64encode(secrets.token_bytes(642)).rstrip(b"=")
|
||||
rotated = base64.urlsafe_b64encode(secrets.token_bytes(642)).rstrip(b"=")
|
||||
self.assertGreater(2 * (len(fresh) + len(rotated)), 2560)
|
||||
requests = []
|
||||
state = {"fail": False}
|
||||
|
||||
class Endpoint(BaseHTTPRequestHandler):
|
||||
def do_POST(self):
|
||||
body = self.rfile.read(int(self.headers.get("Content-Length", "0")))
|
||||
requests.append(parse_qs(body.decode("ascii")))
|
||||
if state["fail"]:
|
||||
self.send_response(401)
|
||||
self.send_header("Content-Length", "0")
|
||||
self.end_headers()
|
||||
return
|
||||
response = json.dumps({"access_token": fresh.decode(), "token_type": "Bearer", "expires_in": 3600, "refresh_token": rotated.decode()}).encode("ascii")
|
||||
self.send_response(200)
|
||||
self.send_header("Content-Type", "application/json")
|
||||
self.send_header("Content-Length", str(len(response)))
|
||||
self.end_headers()
|
||||
self.wfile.write(response)
|
||||
|
||||
def log_message(self, *args):
|
||||
pass
|
||||
|
||||
server = ThreadingHTTPServer(("127.0.0.1", 0), Endpoint)
|
||||
thread = threading.Thread(target=server.serve_forever, daemon=True)
|
||||
thread.start()
|
||||
host = "127.0.0.1:" + str(server.server_port)
|
||||
target_name = vault.target_name(namespace, "http", host, "oauth2")
|
||||
api = ctypes.WinDLL("Advapi32.dll", use_last_error=True)
|
||||
api.CredReadW.argtypes = [wintypes.LPCWSTR, wintypes.DWORD, wintypes.DWORD, ctypes.POINTER(ctypes.c_void_p)]
|
||||
api.CredReadW.restype = wintypes.BOOL
|
||||
api.CredFree.argtypes = [ctypes.c_void_p]
|
||||
api.CredDeleteW.argtypes = [wintypes.LPCWSTR, wintypes.DWORD, wintypes.DWORD]
|
||||
api.CredDeleteW.restype = wintypes.BOOL
|
||||
prior = ctypes.c_void_p()
|
||||
exists = api.CredReadW(target_name, 1, 0, ctypes.byref(prior))
|
||||
if exists:
|
||||
api.CredFree(prior)
|
||||
created = False
|
||||
try:
|
||||
self.assertFalse(exists, "Disposable test target already exists")
|
||||
with tempfile.TemporaryDirectory() as cwd:
|
||||
config = Path(cwd) / "gitconfig"
|
||||
weird_dir = Path(cwd) / "space ' $(literal) `literal`"
|
||||
weird_dir.mkdir()
|
||||
installed = weird_dir / SCRIPT.name
|
||||
shutil.copyfile(SCRIPT, installed)
|
||||
helper = bootstrap.helper_command(sys.executable, str(installed), namespace)
|
||||
env = {k: v for k, v in os.environ.items() if not k.startswith(("GIT_TRACE", "GCM_TRACE")) and k != "GIT_CURL_VERBOSE"}
|
||||
env.update({"GIT_CONFIG_GLOBAL": str(config), "GIT_CONFIG_NOSYSTEM": "1", "XDG_CONFIG_HOME": cwd, "GIT_TERMINAL_PROMPT": "0", "GIT_TRACE": "0", "GIT_TRACE_CURL": "0", "GCM_TRACE": "0"})
|
||||
request = ("protocol=http\nhost=" + host + "\nusername=oauth2\n").encode("ascii")
|
||||
|
||||
def git(*args, input):
|
||||
return subprocess.run(["git", *args], input=input, stdout=subprocess.PIPE, stderr=subprocess.PIPE, cwd=cwd, env=env, timeout=20, check=True).stdout
|
||||
|
||||
def helper_cmd(operation, payload):
|
||||
return subprocess.run([sys.executable, str(installed), "--namespace", namespace, operation], input=payload, stdout=subprocess.PIPE, stderr=subprocess.PIPE, cwd=cwd, env=env, timeout=10, check=True).stdout
|
||||
|
||||
for key, value in [("credential.helper", ""), ("credential.helper", helper),
|
||||
("credential.http://" + host + ".username", "oauth2"),
|
||||
("credential.http://" + host + ".oauthClientId", "fixture-client"),
|
||||
("credential.http://" + host + ".oauthTokenURL", "/token")]:
|
||||
git("config", "--file", str(config), "--add", key, value, input=b"")
|
||||
vault.write_record(target_name, {"username": "oauth2", "password": initial.decode(),
|
||||
"password_expiry_utc": "1", "oauth_refresh_token": initial.decode()})
|
||||
created = True
|
||||
# Several Git processes can encounter expiry at once. The
|
||||
# refresh endpoint rotates a single-use refresh token.
|
||||
with ThreadPoolExecutor(max_workers=4) as pool:
|
||||
results = list(pool.map(lambda _: git("credential", "fill", input=request + b"\n"), range(4)))
|
||||
credential = results[0]
|
||||
for result in results:
|
||||
self.assertIn(b"password=" + fresh, result)
|
||||
fields = dict(line.split(b"=", 1) for line in credential.splitlines() if b"=" in line)
|
||||
self.assertEqual(fields.get(b"password"), fresh)
|
||||
self.assertEqual(fields.get(b"oauth_refresh_token"), rotated)
|
||||
self.assertGreater(int(fields[b"password_expiry_utc"]), int(time.time()))
|
||||
self.assertEqual(requests[0].get("grant_type"), ["refresh_token"])
|
||||
self.assertEqual(requests[0].get("refresh_token"), [initial.decode()])
|
||||
# A later Git operation can fail before approve; the rotated
|
||||
# refresh token must already be persisted at this point.
|
||||
stored = helper_cmd("get", request + b"\n")
|
||||
fields = dict(line.split(b"=", 1) for line in stored.splitlines() if b"=" in line)
|
||||
self.assertEqual(fields.get(b"password"), fresh)
|
||||
self.assertEqual(fields.get(b"oauth_refresh_token"), rotated)
|
||||
git("credential", "approve", input=credential + b"\n")
|
||||
# A delayed approve from another operation must not roll the
|
||||
# vault back to an already-rotated refresh token.
|
||||
helper_cmd("store", request + b"password=" + initial + b"\npassword_expiry_utc=1\noauth_refresh_token=" + initial + b"\n\n")
|
||||
self.assertEqual(helper_cmd("get", request + b"\n"), stored)
|
||||
self.assertIn(b"password=" + fresh, git("credential", "fill", input=request + b"\n"))
|
||||
self.assertEqual(len(requests), 1, "A valid stored access token should be reused")
|
||||
git("credential", "reject", input=credential + b"\n")
|
||||
rejected = vault.read_record(target_name)
|
||||
self.assertEqual(rejected["oauth_refresh_token"], rotated.decode())
|
||||
self.assertTrue(rejected["rejected"], "A rejected access token should be refreshed, not deleted")
|
||||
refreshed = git("credential", "fill", input=request + b"\n")
|
||||
self.assertIn(b"password=" + fresh, refreshed)
|
||||
self.assertEqual(requests[1].get("refresh_token"), [rotated.decode()])
|
||||
git("credential", "reject", input=refreshed + b"\n")
|
||||
state["fail"] = True
|
||||
blocked = subprocess.run(["git", "credential", "fill"], input=request + b"\n", stdout=subprocess.PIPE, stderr=subprocess.PIPE, cwd=cwd, env=env, timeout=10, check=False)
|
||||
self.assertNotEqual(blocked.returncode, 0, "Failed refresh should stop without a browser loop")
|
||||
self.assertEqual(vault.read_record(target_name)["oauth_refresh_token"], rotated.decode())
|
||||
state["fail"] = False
|
||||
self.assertIn(b"password=" + fresh, git("credential", "fill", input=request + b"\n"))
|
||||
self.assertEqual(len(requests), 4)
|
||||
finally:
|
||||
if created and not api.CredDeleteW(target_name, 1, 0) and ctypes.get_last_error() != 1168:
|
||||
self.fail("Could not remove the disposable test credential")
|
||||
server.shutdown()
|
||||
thread.join(timeout=5)
|
||||
server.server_close()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -1,48 +0,0 @@
|
||||
"""Opt-in integration test using only a disposable Windows vault namespace."""
|
||||
import ctypes
|
||||
from ctypes import wintypes
|
||||
import importlib.util
|
||||
import os
|
||||
from pathlib import Path
|
||||
import secrets
|
||||
import sys
|
||||
import tempfile
|
||||
import unittest
|
||||
|
||||
SCRIPT = Path(__file__).resolve().parents[1] / "skills" / "updating-gitea-repositories" / "scripts" / "credential_migrate.py"
|
||||
spec = importlib.util.spec_from_file_location("credential_migrate_vault_test", SCRIPT)
|
||||
module = importlib.util.module_from_spec(spec)
|
||||
spec.loader.exec_module(module)
|
||||
|
||||
|
||||
@unittest.skipUnless(sys.platform == "win32" and os.environ.get("GITEA_TEST_WINDOWS_VAULT") == "1", "opt-in Windows vault test")
|
||||
class WindowsVaultTest(unittest.TestCase):
|
||||
def test_secure_store_readback_and_trace_suppression(self):
|
||||
namespace = "gitea-skill-test-" + secrets.token_hex(16)
|
||||
host = "credential-migration-test.invalid"
|
||||
target_name = namespace + ":https://" + host
|
||||
api = ctypes.WinDLL("Advapi32.dll", use_last_error=True)
|
||||
api.CredDeleteW.argtypes = [wintypes.LPCWSTR, wintypes.DWORD, wintypes.DWORD]
|
||||
api.CredDeleteW.restype = wintypes.BOOL
|
||||
class Source:
|
||||
def get(self, request):
|
||||
return {"username": b"fictional-alice", "password": b"fictional-secret"}
|
||||
module._assert_windows_namespace_empty(namespace)
|
||||
try:
|
||||
with tempfile.TemporaryDirectory() as cwd:
|
||||
trace = Path(cwd) / "trace.log"
|
||||
config = Path(cwd) / "config"
|
||||
config.write_text("[credential]\n trace = " + str(trace).replace("\\", "/") + "\n traceSecrets = true\n", encoding="utf-8")
|
||||
env = module.safe_environment()
|
||||
env.update({"GCM_NAMESPACE": namespace, "GCM_CREDENTIAL_STORE": "wincredman", "GIT_CONFIG_NOSYSTEM": "1", "GIT_CONFIG_GLOBAL": str(config), "XDG_CONFIG_HOME": cwd})
|
||||
target = module.NewWindowsNamespace(["git", "credential-manager"], env, cwd, namespace)
|
||||
module.migrate("https://" + host, "fictional-alice", Source(), target)
|
||||
self.assertFalse(trace.exists() and b"fictional-secret" in trace.read_bytes(), "GCM traced the test credential")
|
||||
finally:
|
||||
if not api.CredDeleteW(target_name, 1, 0) and ctypes.get_last_error() != 1168:
|
||||
self.fail("Could not remove the disposable test credential")
|
||||
module._assert_windows_namespace_empty(namespace)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in new issue
Block a user