wooo/awoooi
1
1
Fork 0
Code Issues 2 Pull Requests 1 Actions 2 Packages Projects Releases Wiki Activity

docs(security): add supply chain contract manifest [skip ci]

Browse Source
This commit is contained in:
Your Name
2026-05-12 23:23:17 +08:00
parent c652f37b69
commit 8ce318ac1c
65 changed files with 8608 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

203
scripts/security/git-remote-refs-probe.py Normal file
View File

@@ -0,0 +1,203 @@
#!/usr/bin/env python3
"""Git remote refs 只讀探測工具。
此工具用 `git ls-remote --heads --tags` 讀取指定本機 repo 的 remote refs，
並比對本機 HEAD / branch。它不 fetch、不 clone、不 push，也不修改 remote。
"""
from __future__ import annotations
import argparse
import json
import subprocess
from pathlib import Path
from urllib.parse import urlsplit, urlunsplit
def redact_url(value: str) -> str:
if "://" not in value:
if "@" in value and ":" in value.split("@", 1)[1]:
return value.split("@", 1)[1]
return value
parts = urlsplit(value)
netloc = parts.netloc.split("@", 1)[-1]
return urlunsplit((parts.scheme, netloc, parts.path, parts.query, parts.fragment))
def run_git(repo: Path, args: list[str], timeout: int) -> subprocess.CompletedProcess[str]:
try:
return subprocess.run(
["git", *args],
cwd=repo,
check=False,
capture_output=True,
text=True,
timeout=timeout,
)
except subprocess.TimeoutExpired:
return subprocess.CompletedProcess(["git", *args], 124, "", "git command timeout")
def git_value(repo: Path, args: list[str], timeout: int) -> str:
result = run_git(repo, args, timeout)
if result.returncode != 0:
return ""
return result.stdout.strip()
def parse_refs(output: str) -> tuple[dict[str, str], dict[str, str]]:
heads: dict[str, str] = {}
tags: dict[str, str] = {}
for line in output.splitlines():
parts = line.split()
if len(parts) != 2:
continue
sha, ref = parts
if ref.startswith("refs/heads/"):
heads[ref.removeprefix("refs/heads/")] = sha
elif ref.startswith("refs/tags/"):
tag = ref.removeprefix("refs/tags/").removesuffix("^{}")
tags[tag] = sha
return heads, tags
def parse_repo_arg(value: str) -> tuple[str, Path, str]:
parts = value.split("=")
if len(parts) not in (2, 3):
raise argparse.ArgumentTypeError("--repo 必須是 label=/absolute/path 或 label=/absolute/path=remote")
label = parts[0].strip()
path = Path(parts[1].strip()).expanduser().resolve()
remote = parts[2].strip() if len(parts) == 3 else "origin"
if not label or not str(path) or not remote:
raise argparse.ArgumentTypeError("--repo label/path/remote 不可為空")
return label, path, remote
def probe_repo(label: str, repo: Path, remote: str, timeout: int) -> dict[str, object]:
local_head = git_value(repo, ["rev-parse", "HEAD"], timeout)
local_branch = git_value(repo, ["branch", "--show-current"], timeout)
remote_url = git_value(repo, ["remote", "get-url", remote], timeout)
ls_remote = run_git(repo, ["ls-remote", "--heads", "--tags", remote], timeout)
if ls_remote.returncode == 0:
heads, tags = parse_refs(ls_remote.stdout)
remote_branch_sha = heads.get(local_branch, "")
if local_branch and remote_branch_sha and remote_branch_sha == local_head:
status = "aligned_current_branch"
elif heads:
status = "reachable_drift_or_unknown"
else:
status = "reachable_no_heads"
error_summary = ""
else:
heads = {}
tags = {}
remote_branch_sha = ""
status = "unreachable"
stderr = ls_remote.stderr.strip()
if "Permission denied" in stderr:
error_summary = "SSH 權限不足或 remote 不可讀"
elif "Repository not found" in stderr:
error_summary = "remote repo not found 或未授權"
else:
tail = (stderr or ls_remote.stdout.strip()).splitlines()[-1:]
error_summary = tail[0] if tail else "git ls-remote failed"
return {
"label": label,
"repo_path": str(repo),
"remote": remote,
"remote_url_redacted": redact_url(remote_url),
"status": status,
"local_branch": local_branch,
"local_head": local_head,
"remote_current_branch_sha": remote_branch_sha,
"head_count": len(heads),
"tag_count": len(tags),
"heads": [{"name": name, "sha": sha} for name, sha in sorted(heads.items())],
"tags": [{"name": name, "sha": sha} for name, sha in sorted(tags.items())],
"error_summary": error_summary,
}
def build_payload(group_name: str, repo_args: list[tuple[str, Path, str]], timeout: int) -> dict[str, object]:
repos = [probe_repo(label, path, remote, timeout) for label, path, remote in repo_args]
aligned = sum(1 for repo in repos if repo["status"] == "aligned_current_branch")
unreachable = sum(1 for repo in repos if repo["status"] == "unreachable")
return {
"schema_version": "git_remote_refs_probe_v1",
"group_name": group_name,
"status": "ok" if unreachable == 0 else "partial",
"repo_count": len(repos),
"aligned_current_branch_count": aligned,
"unreachable_count": unreachable,
"repos": repos,
}
def write_markdown(payload: dict[str, object], path: Path) -> None:
lines = [
"# Git Remote Refs Probe 快照",
"",
"| 項目 | 值 |",
"|------|----|",
f"| 群組 | `{payload['group_name']}` |",
f"| 狀態 | `{payload['status']}` |",
f"| repo 數 | `{payload['repo_count']}` |",
f"| aligned current branch | `{payload['aligned_current_branch_count']}` |",
f"| unreachable | `{payload['unreachable_count']}` |",
"",
"## Repo refs",
"",
"| Label | Remote URL | Status | Local branch | Local HEAD | Remote branch SHA | Heads | Tags | Error |",
"|-------|------------|--------|--------------|------------|-------------------|-------|------|-------|",
]
for repo in payload.get("repos", []):
if not isinstance(repo, dict):
continue
lines.append(
"| "
+ " | ".join(
[
f"`{repo.get('label', '')}`",
f"`{repo.get('remote_url_redacted', '')}`",
f"`{repo.get('status', '')}`",
f"`{repo.get('local_branch', '')}`",
f"`{str(repo.get('local_head', ''))[:7]}`",
f"`{str(repo.get('remote_current_branch_sha', ''))[:7]}`",
f"`{repo.get('head_count', 0)}`",
f"`{repo.get('tag_count', 0)}`",
str(repo.get("error_summary", "") or "無"),
]
)
+ " |"
)
lines.extend(
[
"",
"> 注意：本檔只使用 `git ls-remote` 做 read-only refs 探測；未 fetch、未 clone、未 push。",
"",
]
)
path.write_text("\n".join(lines), encoding="utf-8")
def main() -> int:
parser = argparse.ArgumentParser()
parser.add_argument("--group-name", required=True)
parser.add_argument("--repo", action="append", type=parse_repo_arg, required=True)
parser.add_argument("--output-json", required=True)
parser.add_argument("--output-md", required=True)
parser.add_argument("--timeout", type=int, default=10)
args = parser.parse_args()
payload = build_payload(args.group_name, args.repo, args.timeout)
Path(args.output_json).write_text(
json.dumps(payload, ensure_ascii=False, indent=2) + "\n",
encoding="utf-8",
)
write_markdown(payload, Path(args.output_md))
return 0
if __name__ == "__main__":
raise SystemExit(main())

255
scripts/security/gitea-repo-inventory.py Normal file
View File

@@ -0,0 +1,255 @@
#!/usr/bin/env python3
"""Read-only Gitea repo 全量盤點工具。
此工具可查詢 Gitea org/user API endpoint，或吃管理介面匯出的 JSON。
執行期間不寫入 Gitea，也不會把 API token 寫入輸出檔。
"""
from __future__ import annotations
import argparse
import json
import os
import sys
import urllib.error
import urllib.request
from pathlib import Path
from urllib.parse import quote, urlencode, urlsplit, urlunsplit
def redact_url(value: str) -> str:
if "://" not in value:
if "@" in value and ":" in value.split("@", 1)[1]:
return value.split("@", 1)[1]
return value
parts = urlsplit(value)
netloc = parts.netloc.split("@", 1)[-1]
return urlunsplit((parts.scheme, netloc, parts.path, parts.query, parts.fragment))
def api_get_json(url: str, token: str | None, timeout: int) -> tuple[int, object]:
headers = {
"Accept": "application/json",
"User-Agent": "awoooi-security-inventory/1.0",
}
if token:
headers["Authorization"] = f"token {token}"
request = urllib.request.Request(url, headers=headers, method="GET")
try:
with urllib.request.urlopen(request, timeout=timeout) as response:
body = response.read().decode("utf-8")
return response.status, json.loads(body)
except urllib.error.HTTPError as exc:
body = exc.read().decode("utf-8", errors="replace")
try:
payload: object = json.loads(body)
except json.JSONDecodeError:
payload = {"message": body.strip()}
return exc.code, payload
def repo_summary(repo: dict[str, object], github_owner: str | None) -> dict[str, object]:
owner = repo.get("owner") if isinstance(repo.get("owner"), dict) else {}
owner_name = owner.get("login") if isinstance(owner, dict) else None
raw_full_name = str(repo.get("full_name") or "")
name = str(repo.get("name") or raw_full_name.rsplit("/", 1)[-1] or "")
full_name = raw_full_name or (f"{owner_name}/{name}" if owner_name else name)
clone_url = str(repo.get("clone_url") or repo.get("html_url") or "")
ssh_url = str(repo.get("ssh_url") or "")
github_repo = f"{github_owner}/{name}" if github_owner and name else ""
return {
"gitea_repo": full_name,
"name": name,
"owner": owner_name or "",
"private": bool(repo.get("private", False)),
"empty": bool(repo.get("empty", False)),
"archived": bool(repo.get("archived", False)),
"default_branch": str(repo.get("default_branch") or ""),
"clone_url_redacted": redact_url(clone_url),
"ssh_url_redacted": redact_url(ssh_url),
"github_repo_candidate": github_repo,
}
def load_export(path: Path) -> object:
return json.loads(path.read_text(encoding="utf-8"))
def extract_repos(payload: object) -> list[dict[str, object]]:
if isinstance(payload, list):
return [item for item in payload if isinstance(item, dict)]
if isinstance(payload, dict):
for key in ("data", "repos", "repositories"):
value = payload.get(key)
if isinstance(value, list):
return [item for item in value if isinstance(item, dict)]
return []
def build_inventory(
*,
base_url: str,
org: str,
github_owner: str | None,
token_present: bool,
http_status: int | None,
payload: object,
query_mode: str,
query: str,
) -> dict[str, object]:
repos = [repo_summary(repo, github_owner) for repo in extract_repos(payload)]
if query_mode == "export":
visibility_scope = "admin_export"
elif token_present:
visibility_scope = "authenticated"
else:
visibility_scope = "public_only"
if repos and visibility_scope in ("authenticated", "admin_export"):
status = "ok"
elif repos:
status = "partial"
else:
status = "blocked"
blocking_reason = ""
if repos and status == "partial":
blocking_reason = "未提供 token，結果只代表公開可見 repo；private/internal repos 仍需只讀 token 或管理匯出"
elif not repos:
if http_status in (401, 403):
blocking_reason = "Gitea API 需要只讀 token 或權限不足"
elif query_mode == "search" and http_status == 200:
blocking_reason = "Gitea public repo search 未回傳 repo，可能沒有公開 repo 或需要只讀 token"
elif http_status == 404:
blocking_reason = "Gitea API 查無 org/user repos，需確認 org 名稱或使用管理匯出"
elif http_status is None:
blocking_reason = "匯入檔案沒有可解析的 repo list"
else:
blocking_reason = f"Gitea API 回應無 repo list，HTTP {http_status}"
return {
"schema_version": "gitea_repo_inventory_v1",
"base_url": redact_url(base_url.rstrip("/")),
"org": org,
"github_owner": github_owner or "",
"query_mode": query_mode,
"query": query,
"visibility_scope": visibility_scope,
"token_present": token_present,
"http_status": http_status,
"status": status,
"blocking_reason": blocking_reason,
"repo_count": len(repos),
"repos": repos,
}
def write_markdown(inventory: dict[str, object], path: Path) -> None:
lines = [
"# Gitea Repo 全量盤點快照",
"",
"| 項目 | 值 |",
"|------|----|",
f"| 狀態 | `{inventory['status']}` |",
f"| Gitea base URL | `{inventory['base_url']}` |",
f"| Org/User | `{inventory['org']}` |",
f"| GitHub owner 候選 | `{inventory['github_owner']}` |",
f"| 查詢模式 | `{inventory.get('query_mode', '')}` |",
f"| 查詢字串 | `{inventory.get('query', '')}` |",
f"| 可見性範圍 | `{inventory.get('visibility_scope', '')}` |",
f"| 是否提供 token | `{inventory['token_present']}` |",
f"| HTTP status | `{inventory['http_status']}` |",
f"| Repo 數量 | `{inventory['repo_count']}` |",
f"| 阻塞原因 | {inventory['blocking_reason'] or '無'} |",
"",
"## Repo 清單",
"",
"| Gitea repo | GitHub 候選 | default branch | private | archived |",
"|------------|------------------|----------------|---------|----------|",
]
repos = inventory.get("repos")
if isinstance(repos, list):
for repo in repos:
if not isinstance(repo, dict):
continue
lines.append(
"| "
+ " | ".join(
[
f"`{repo.get('gitea_repo', '')}`",
f"`{repo.get('github_repo_candidate', '')}`",
f"`{repo.get('default_branch', '')}`",
f"`{repo.get('private', '')}`",
f"`{repo.get('archived', '')}`",
]
)
+ " |"
)
lines.extend(
[
"",
"> 注意：本檔由 read-only Gitea inventory 工具產生，不包含 API token 或 remote URL 帳密。",
"",
]
)
path.write_text("\n".join(lines), encoding="utf-8")
def main() -> int:
parser = argparse.ArgumentParser()
parser.add_argument("--base-url", default="http://192.168.0.110:3001")
parser.add_argument("--org", default="wooo")
parser.add_argument("--scope", choices=["org", "user", "search"], default="org")
parser.add_argument("--query", default="")
parser.add_argument("--limit", type=int, default=50)
parser.add_argument("--github-owner", default="")
parser.add_argument("--token-env", default="GITEA_READONLY_TOKEN")
parser.add_argument("--input-json")
parser.add_argument("--output-json", required=True)
parser.add_argument("--output-md", required=True)
parser.add_argument("--timeout", type=int, default=5)
args = parser.parse_args()
token = os.environ.get(args.token_env)
http_status: int | None = None
if args.input_json:
payload = load_export(Path(args.input_json))
query_mode = "export"
query = "input-json"
else:
if args.scope == "search":
params = {"limit": str(args.limit)}
if args.query:
params["q"] = args.query
url = f"{args.base_url.rstrip('/')}/api/v1/repos/search?{urlencode(params)}"
else:
quoted = quote(args.org, safe="")
prefix = "orgs" if args.scope == "org" else "users"
url = f"{args.base_url.rstrip('/')}/api/v1/{prefix}/{quoted}/repos"
http_status, payload = api_get_json(url, token, args.timeout)
query_mode = args.scope
query = args.query
inventory = build_inventory(
base_url=args.base_url,
org=args.org,
github_owner=args.github_owner or None,
token_present=bool(token),
http_status=http_status,
payload=payload,
query_mode=query_mode,
query=query,
)
Path(args.output_json).write_text(
json.dumps(inventory, ensure_ascii=False, indent=2) + "\n",
encoding="utf-8",
)
write_markdown(inventory, Path(args.output_md))
if inventory["status"] == "blocked":
print(inventory["blocking_reason"], file=sys.stderr)
return 2
return 0
if __name__ == "__main__":
raise SystemExit(main())

138
scripts/security/github-target-probe.py Normal file
View File

@@ -0,0 +1,138 @@
#!/usr/bin/env python3
"""GitHub target repo 只讀存在性探測。
此工具使用 `git ls-remote --heads` 檢查候選 GitHub repo 是否可讀。
它不 clone、不 fetch、不 push，也不寫入任何 remote。
"""
from __future__ import annotations
import argparse
import json
import subprocess
from pathlib import Path
def probe_candidate(candidate: str, timeout: int) -> dict[str, object]:
url = f"https://github.com/{candidate}.git"
try:
result = subprocess.run(
["git", "ls-remote", "--heads", url],
check=False,
capture_output=True,
text=True,
timeout=timeout,
)
except subprocess.TimeoutExpired:
return {
"github_repo": candidate,
"url_redacted": url,
"status": "timeout",
"head_count": 0,
"heads": [],
"error_summary": "git ls-remote timeout",
}
heads = []
if result.returncode == 0:
for line in result.stdout.splitlines():
parts = line.split()
if len(parts) != 2 or not parts[1].startswith("refs/heads/"):
continue
heads.append(
{
"name": parts[1].removeprefix("refs/heads/"),
"sha": parts[0],
}
)
status = "exists" if heads else "exists_empty_or_no_heads"
error_summary = ""
else:
stderr = result.stderr.strip()
if "Repository not found" in stderr:
status = "not_found_or_private"
error_summary = "GitHub 回應 repository not found；可能未建立或為 private 且未授權"
else:
status = "error"
error_summary = stderr.splitlines()[-1] if stderr else "git ls-remote failed"
return {
"github_repo": candidate,
"url_redacted": url,
"status": status,
"head_count": len(heads),
"heads": heads,
"error_summary": error_summary,
}
def write_markdown(payload: dict[str, object], path: Path) -> None:
lines = [
"# GitHub Target Probe 快照",
"",
"| 項目 | 值 |",
"|------|----|",
f"| 狀態 | `{payload['status']}` |",
f"| 候選 repo 數 | `{payload['candidate_count']}` |",
f"| exists | `{payload['exists_count']}` |",
f"| not found or private | `{payload['not_found_or_private_count']}` |",
"",
"## 候選 Repo",
"",
"| GitHub repo | status | heads | error |",
"|-------------|--------|-------|-------|",
]
for candidate in payload.get("candidates", []):
if not isinstance(candidate, dict):
continue
lines.append(
"| "
+ " | ".join(
[
f"`{candidate.get('github_repo', '')}`",
f"`{candidate.get('status', '')}`",
f"`{candidate.get('head_count', 0)}`",
str(candidate.get("error_summary", "") or "無"),
]
)
+ " |"
)
lines.extend(
[
"",
"> 注意：`not_found_or_private` 只代表未授權 read-only probe 看不到，不等同確認不存在。",
"",
]
)
path.write_text("\n".join(lines), encoding="utf-8")
def main() -> int:
parser = argparse.ArgumentParser()
parser.add_argument("--candidate", action="append", required=True)
parser.add_argument("--output-json", required=True)
parser.add_argument("--output-md", required=True)
parser.add_argument("--timeout", type=int, default=10)
args = parser.parse_args()
candidates = [probe_candidate(candidate, args.timeout) for candidate in args.candidate]
exists_count = sum(1 for item in candidates if item["status"].startswith("exists"))
not_found_count = sum(1 for item in candidates if item["status"] == "not_found_or_private")
payload = {
"schema_version": "github_target_probe_v1",
"status": "ok",
"candidate_count": len(candidates),
"exists_count": exists_count,
"not_found_or_private_count": not_found_count,
"candidates": candidates,
}
Path(args.output_json).write_text(
json.dumps(payload, ensure_ascii=False, indent=2) + "\n",
encoding="utf-8",
)
write_markdown(payload, Path(args.output_md))
return 0
if __name__ == "__main__":
raise SystemExit(main())

352
scripts/security/local-git-remote-inventory.py Normal file
View File

@@ -0,0 +1,352 @@
#!/usr/bin/env python3
"""本機 Git remote 只讀盤點工具。
此工具掃描指定 root 底下可見的 Git working tree，讀取 `.git/config`
中的 remote URL，並在輸出前移除 URL 內的帳密。它不會 fetch、push、
修改 remote，也不會連線到 GitHub 或 Gitea。
"""
from __future__ import annotations
import argparse
import configparser
import json
import os
import sys
from pathlib import Path
from urllib.parse import urlsplit, urlunsplit
DEFAULT_EXCLUDE_NAMES = {
".cache",
".cargo",
".claude",
".codex",
".gemini",
".git",
".gradle",
".npm",
".nvm",
".openclaw",
".pyenv",
".rustup",
".Trash",
".venv",
"__pycache__",
"Applications",
"Applications (Parallels)",
"Caches",
"DerivedData",
"Library",
"Movies",
"Music",
"node_modules",
"Parallels",
"Pictures",
"venv",
}
def redact_url(value: str) -> str:
if "://" not in value:
if "@" in value and ":" in value.split("@", 1)[1]:
return value.split("@", 1)[1]
return value
parts = urlsplit(value)
netloc = parts.netloc.split("@", 1)[-1]
return urlunsplit((parts.scheme, netloc, parts.path, parts.query, parts.fragment))
def repo_slug_from_url(value: str) -> str:
redacted = redact_url(value).removesuffix("/")
if "://" in redacted:
path = urlsplit(redacted).path.strip("/")
elif ":" in redacted:
path = redacted.split(":", 1)[1].strip("/")
else:
path = redacted.strip("/")
return path.removesuffix(".git")
def classify_remote(url: str, gitea_fragment: str) -> str:
lowered = url.lower()
if gitea_fragment.lower() in lowered:
return "gitea"
if "github.com" in lowered:
return "github"
if "192.168.0.110:8929" in lowered:
return "gitlab_110"
if "192.168.0.110" in lowered:
return "internal_git_110"
return "other"
def git_config_path(repo_path: Path) -> Path | None:
git_path = repo_path / ".git"
if git_path.is_dir():
config_path = git_path / "config"
return config_path if config_path.exists() else None
if not git_path.is_file():
return None
text = git_path.read_text(encoding="utf-8", errors="replace")
for line in text.splitlines():
if line.startswith("gitdir:"):
raw_gitdir = line.split(":", 1)[1].strip()
gitdir = Path(raw_gitdir)
if not gitdir.is_absolute():
gitdir = (repo_path / gitdir).resolve()
config_path = gitdir / "config"
return config_path if config_path.exists() else None
return None
def remote_name(section: str) -> str | None:
prefix = 'remote "'
if section.startswith(prefix) and section.endswith('"'):
return section[len(prefix) : -1]
return None
def read_remotes(repo_path: Path, gitea_fragment: str) -> list[dict[str, str]]:
config_path = git_config_path(repo_path)
if config_path is None:
return []
parser = configparser.RawConfigParser(strict=False)
parser.read(config_path, encoding="utf-8")
remotes: list[dict[str, str]] = []
for section in parser.sections():
name = remote_name(section)
if not name or not parser.has_option(section, "url"):
continue
raw_url = parser.get(section, "url").strip()
redacted_url = redact_url(raw_url)
remotes.append(
{
"name": name,
"kind": classify_remote(redacted_url, gitea_fragment),
"url_redacted": redacted_url,
"repo_slug": repo_slug_from_url(redacted_url),
}
)
return remotes
def should_skip_dir(path: Path, root: Path, max_depth: int, exclude_names: set[str]) -> bool:
if path.name in exclude_names:
return True
try:
depth = len(path.relative_to(root).parts)
except ValueError:
return True
return depth > max_depth
def find_repos(roots: list[Path], max_depth: int, exclude_names: set[str]) -> list[Path]:
repos: dict[str, Path] = {}
for root in roots:
if not root.exists():
continue
for current, dirs, _files in os.walk(root):
current_path = Path(current)
if should_skip_dir(current_path, root, max_depth, exclude_names):
dirs[:] = []
continue
if (current_path / ".git").exists():
repos[str(current_path.resolve())] = current_path.resolve()
dirs[:] = []
continue
dirs[:] = [
name
for name in dirs
if not should_skip_dir(current_path / name, root, max_depth, exclude_names)
]
return sorted(repos.values(), key=lambda path: str(path))
def summarize_repo(repo_path: Path, remotes: list[dict[str, str]]) -> dict[str, object]:
gitea = [remote["repo_slug"] for remote in remotes if remote["kind"] == "gitea"]
github = [remote["repo_slug"] for remote in remotes if remote["kind"] == "github"]
internal_110 = [
remote["repo_slug"]
for remote in remotes
if remote["kind"] in ("internal_git_110", "gitlab_110")
]
if gitea and github:
status = "mapped"
elif gitea:
status = "gitea_only_local"
elif github:
status = "github_only_local"
elif internal_110:
status = "internal_110_only"
else:
status = "other_remote"
return {
"repo_path": str(repo_path),
"repo_name": repo_path.name,
"status": status,
"gitea_repos": sorted(set(gitea)),
"github_repos": sorted(set(github)),
"internal_110_repos": sorted(set(internal_110)),
"remotes": remotes,
}
def build_inventory(
roots: list[Path],
max_depth: int,
exclude_names: set[str],
gitea_fragment: str,
) -> dict[str, object]:
repo_paths = find_repos(roots, max_depth, exclude_names)
repos = [
summarize_repo(repo_path, read_remotes(repo_path, gitea_fragment))
for repo_path in repo_paths
]
gitea_linked = [repo for repo in repos if repo["gitea_repos"]]
github_linked = [repo for repo in repos if repo["github_repos"]]
mapped = [repo for repo in repos if repo["status"] == "mapped"]
gitea_only = [repo for repo in repos if repo["status"] == "gitea_only_local"]
github_only = [repo for repo in repos if repo["status"] == "github_only_local"]
internal_110 = [repo for repo in repos if repo["status"] == "internal_110_only"]
unique_gitea = sorted(
{
item
for repo in repos
for item in repo.get("gitea_repos", [])
if isinstance(item, str)
}
)
unique_github = sorted(
{
item
for repo in repos
for item in repo.get("github_repos", [])
if isinstance(item, str)
}
)
unique_internal_110 = sorted(
{
item
for repo in repos
for item in repo.get("internal_110_repos", [])
if isinstance(item, str)
}
)
return {
"schema_version": "local_git_remote_inventory_v1",
"status": "partial" if repos else "empty",
"roots": [str(root) for root in roots],
"max_depth": max_depth,
"gitea_host_fragment": gitea_fragment,
"repo_count": len(repos),
"gitea_linked_count": len(gitea_linked),
"github_linked_count": len(github_linked),
"mapped_count": len(mapped),
"gitea_only_count": len(gitea_only),
"github_only_count": len(github_only),
"internal_110_only_count": len(internal_110),
"unique_gitea_repo_count": len(unique_gitea),
"unique_github_repo_count": len(unique_github),
"unique_internal_110_repo_count": len(unique_internal_110),
"unique_gitea_repos": unique_gitea,
"unique_github_repos": unique_github,
"unique_internal_110_repos": unique_internal_110,
"repos": repos,
}
def write_markdown(inventory: dict[str, object], path: Path) -> None:
lines = [
"# 本機 Git Remote 盤點快照",
"",
"| 項目 | 值 |",
"|------|----|",
f"| 狀態 | `{inventory['status']}` |",
f"| 掃描 root | `{', '.join(inventory['roots'])}` |",
f"| max depth | `{inventory['max_depth']}` |",
f"| Gitea host fragment | `{inventory['gitea_host_fragment']}` |",
f"| repo 數量 | `{inventory['repo_count']}` |",
f"| Gitea linked | `{inventory['gitea_linked_count']}` |",
f"| GitHub linked | `{inventory['github_linked_count']}` |",
f"| mapped | `{inventory['mapped_count']}` |",
f"| Gitea-only local | `{inventory['gitea_only_count']}` |",
f"| GitHub-only local | `{inventory['github_only_count']}` |",
f"| Internal 110-only local | `{inventory['internal_110_only_count']}` |",
f"| 去重後 Gitea repo | `{inventory['unique_gitea_repo_count']}` |",
f"| 去重後 GitHub repo | `{inventory['unique_github_repo_count']}` |",
f"| 去重後 110 內部 repo | `{inventory['unique_internal_110_repo_count']}` |",
"",
"## Repo 對照",
"",
"| 狀態 | 本機路徑 | Gitea repo | GitHub repo | 110 內部 remote |",
"|------|----------|------------|-------------|----------------|",
]
repos = inventory.get("repos")
if isinstance(repos, list):
for repo in repos:
if not isinstance(repo, dict):
continue
gitea = ", ".join(f"`{item}`" for item in repo.get("gitea_repos", [])) or "-"
github = ", ".join(f"`{item}`" for item in repo.get("github_repos", [])) or "-"
internal_110 = (
", ".join(f"`{item}`" for item in repo.get("internal_110_repos", [])) or "-"
)
lines.append(
"| "
+ " | ".join(
[
f"`{repo.get('status', '')}`",
f"`{repo.get('repo_path', '')}`",
gitea,
github,
internal_110,
]
)
+ " |"
)
lines.extend(
[
"",
"> 注意：本檔只代表本機指定 roots 可見的 Git working tree，不等同 Gitea server 全量 repo 清單。",
"> 輸出前已移除 remote URL 中的 username、password、token。",
"",
]
)
path.write_text("\n".join(lines), encoding="utf-8")
def main() -> int:
parser = argparse.ArgumentParser()
parser.add_argument("--root", action="append", required=True)
parser.add_argument("--max-depth", type=int, default=4)
parser.add_argument("--exclude-name", action="append", default=[])
parser.add_argument("--gitea-host-fragment", default="192.168.0.110:3001")
parser.add_argument("--output-json", required=True)
parser.add_argument("--output-md", required=True)
args = parser.parse_args()
roots = [Path(root).expanduser().resolve() for root in args.root]
exclude_names = set(DEFAULT_EXCLUDE_NAMES)
exclude_names.update(args.exclude_name)
inventory = build_inventory(
roots=roots,
max_depth=args.max_depth,
exclude_names=exclude_names,
gitea_fragment=args.gitea_host_fragment,
)
Path(args.output_json).write_text(
json.dumps(inventory, ensure_ascii=False, indent=2) + "\n",
encoding="utf-8",
)
write_markdown(inventory, Path(args.output_md))
if inventory["status"] == "empty":
print("沒有找到本機 Git working tree", file=sys.stderr)
return 2
return 0
if __name__ == "__main__":
raise SystemExit(main())

307
scripts/security/local-repo-canonical-probe.py Normal file
View File

@@ -0,0 +1,307 @@
#!/usr/bin/env python3
"""本機 repo canonical lineage 只讀探測。
此工具比較多個本機 Git working tree 的 HEAD、branch、remote 與近期
commit ancestry，協助判斷它們是否可能屬於同一個 canonical repo。
它不 fetch、不push、不修改 remote，也不讀取 commit message。
"""
from __future__ import annotations
import argparse
import configparser
import json
import subprocess
from pathlib import Path
from urllib.parse import urlsplit, urlunsplit
def redact_url(value: str) -> str:
if "://" not in value:
if "@" in value and ":" in value.split("@", 1)[1]:
return value.split("@", 1)[1]
return value
parts = urlsplit(value)
netloc = parts.netloc.split("@", 1)[-1]
return urlunsplit((parts.scheme, netloc, parts.path, parts.query, parts.fragment))
def run_git(repo: Path, args: list[str], timeout: int) -> subprocess.CompletedProcess[str]:
try:
return subprocess.run(
["git", *args],
cwd=repo,
check=False,
capture_output=True,
text=True,
timeout=timeout,
)
except subprocess.TimeoutExpired:
return subprocess.CompletedProcess(["git", *args], 124, "", "git command timeout")
def git_value(repo: Path, args: list[str], timeout: int) -> str:
result = run_git(repo, args, timeout)
if result.returncode != 0:
return ""
return result.stdout.strip()
def git_config_path(repo_path: Path) -> Path | None:
git_path = repo_path / ".git"
if git_path.is_dir():
config_path = git_path / "config"
return config_path if config_path.exists() else None
if not git_path.is_file():
return None
text = git_path.read_text(encoding="utf-8", errors="replace")
for line in text.splitlines():
if line.startswith("gitdir:"):
raw_gitdir = line.split(":", 1)[1].strip()
gitdir = Path(raw_gitdir)
if not gitdir.is_absolute():
gitdir = (repo_path / gitdir).resolve()
config_path = gitdir / "config"
return config_path if config_path.exists() else None
return None
def remote_name(section: str) -> str | None:
prefix = 'remote "'
if section.startswith(prefix) and section.endswith('"'):
return section[len(prefix) : -1]
return None
def read_remotes(repo_path: Path) -> list[dict[str, str]]:
config_path = git_config_path(repo_path)
if config_path is None:
return []
parser = configparser.RawConfigParser(strict=False)
parser.read(config_path, encoding="utf-8")
remotes: list[dict[str, str]] = []
for section in parser.sections():
name = remote_name(section)
if not name or not parser.has_option(section, "url"):
continue
remotes.append(
{
"name": name,
"url_redacted": redact_url(parser.get(section, "url").strip()),
}
)
return remotes
def parse_repo_arg(value: str) -> tuple[str, Path]:
if "=" not in value:
raise argparse.ArgumentTypeError("--repo 必須是 label=/absolute/path")
label, raw_path = value.split("=", 1)
if not label.strip() or not raw_path.strip():
raise argparse.ArgumentTypeError("--repo label 與 path 不可為空")
return label.strip(), Path(raw_path).expanduser().resolve()
def repo_summary(label: str, repo_path: Path, sample_limit: int, git_timeout: int) -> dict[str, object]:
exists = (repo_path / ".git").exists()
if not exists:
return {
"label": label,
"repo_path": str(repo_path),
"exists": False,
"head_sha": "",
"head_short": "",
"branch": "",
"commit_sample_count": 0,
"commits": [],
"remotes": [],
"probe_error": "repo missing",
}
probe_errors: list[str] = []
head_result = run_git(repo_path, ["rev-parse", "HEAD"], git_timeout)
head_sha = head_result.stdout.strip() if head_result.returncode == 0 else ""
if head_result.returncode != 0:
probe_errors.append("HEAD 讀取失敗或逾時")
branch = git_value(repo_path, ["rev-parse", "--abbrev-ref", "HEAD"], git_timeout)
rev_list_result = run_git(repo_path, ["rev-list", f"--max-count={sample_limit}", "HEAD"], git_timeout)
commits = rev_list_result.stdout.splitlines() if rev_list_result.returncode == 0 else []
if rev_list_result.returncode != 0:
probe_errors.append("rev-list 讀取失敗或逾時")
return {
"label": label,
"repo_path": str(repo_path),
"exists": True,
"head_sha": head_sha,
"head_short": head_sha[:7],
"branch": branch,
"commit_sample_count": len(commits),
"commits": commits,
"remotes": read_remotes(repo_path),
"probe_error": "；".join(probe_errors),
}
def compare_repos(left: dict[str, object], right: dict[str, object]) -> dict[str, object]:
left_commits = set(left.get("commits", []))
right_commits = set(right.get("commits", []))
left_head = str(left.get("head_sha") or "")
right_head = str(right.get("head_sha") or "")
common = sorted(left_commits & right_commits)
if not left.get("exists") or not right.get("exists"):
relation = "missing_repo"
elif left_head and left_head == right_head:
relation = "same_head"
elif left.get("probe_error") or right.get("probe_error"):
relation = "partial_probe"
elif right_head and right_head in left_commits:
relation = "left_descends_from_right"
elif left_head and left_head in right_commits:
relation = "right_descends_from_left"
elif common:
relation = "shared_history"
else:
relation = "no_shared_history"
return {
"left_label": left["label"],
"right_label": right["label"],
"relation": relation,
"left_head": left_head,
"right_head": right_head,
"common_commit_count": len(common),
"common_commit_samples": common[:5],
}
def build_payload(
group_name: str,
repo_args: list[tuple[str, Path]],
sample_limit: int,
git_timeout: int,
) -> dict[str, object]:
repos = [repo_summary(label, path, sample_limit, git_timeout) for label, path in repo_args]
comparisons = []
for left_index, left in enumerate(repos):
for right in repos[left_index + 1 :]:
comparisons.append(compare_repos(left, right))
partial = any(item["relation"] == "partial_probe" for item in comparisons)
related = any(
item["relation"]
in ("same_head", "left_descends_from_right", "right_descends_from_left", "shared_history")
for item in comparisons
)
no_shared = any(item["relation"] == "no_shared_history" for item in comparisons)
if partial:
status = "partial"
elif related and no_shared:
status = "mixed"
elif related:
status = "related"
elif comparisons:
status = "unrelated"
else:
status = "partial"
return {
"schema_version": "local_repo_canonical_probe_v1",
"group_name": group_name,
"status": status,
"sample_limit": sample_limit,
"git_timeout_seconds": git_timeout,
"repo_count": len(repos),
"comparison_count": len(comparisons),
"repos": repos,
"comparisons": comparisons,
}
def write_markdown(payload: dict[str, object], path: Path) -> None:
lines = [
"# 本機 Repo Canonical Lineage Probe 快照",
"",
"| 項目 | 值 |",
"|------|----|",
f"| 群組 | `{payload['group_name']}` |",
f"| 狀態 | `{payload['status']}` |",
f"| repo 數 | `{payload['repo_count']}` |",
f"| 比對數 | `{payload['comparison_count']}` |",
f"| sample limit | `{payload['sample_limit']}` |",
f"| git timeout seconds | `{payload['git_timeout_seconds']}` |",
"",
"## Repo HEAD",
"",
"| Label | Path | Branch | HEAD | Remotes |",
"|-------|------|--------|------|---------|",
]
for repo in payload.get("repos", []):
if not isinstance(repo, dict):
continue
remotes = repo.get("remotes", [])
remote_text = ", ".join(
f"`{remote.get('name', '')}:{remote.get('url_redacted', '')}`"
for remote in remotes
if isinstance(remote, dict)
)
lines.append(
"| "
+ " | ".join(
[
f"`{repo.get('label', '')}`",
f"`{repo.get('repo_path', '')}`",
f"`{repo.get('branch', '')}`",
f"`{repo.get('head_short', '')}`",
remote_text or "-",
]
)
+ " |"
)
lines.extend(["", "## Lineage 比對", "", "| Left | Right | Relation | Common commits |", "|------|-------|----------|----------------|"])
for comparison in payload.get("comparisons", []):
if not isinstance(comparison, dict):
continue
lines.append(
"| "
+ " | ".join(
[
f"`{comparison.get('left_label', '')}`",
f"`{comparison.get('right_label', '')}`",
f"`{comparison.get('relation', '')}`",
f"`{comparison.get('common_commit_count', 0)}`",
]
)
+ " |"
)
lines.extend(
[
"",
"> 注意：本檔只比較本機 Git 物件，未 fetch 遠端；common commit sample 只用 SHA，不含 commit message。",
"",
]
)
path.write_text("\n".join(lines), encoding="utf-8")
def main() -> int:
parser = argparse.ArgumentParser()
parser.add_argument("--group-name", required=True)
parser.add_argument("--repo", action="append", type=parse_repo_arg, required=True)
parser.add_argument("--sample-limit", type=int, default=5000)
parser.add_argument("--git-timeout", type=int, default=10)
parser.add_argument("--output-json", required=True)
parser.add_argument("--output-md", required=True)
args = parser.parse_args()
payload = build_payload(args.group_name, args.repo, args.sample_limit, args.git_timeout)
Path(args.output_json).write_text(
json.dumps(payload, ensure_ascii=False, indent=2) + "\n",
encoding="utf-8",
)
write_markdown(payload, Path(args.output_md))
return 0
if __name__ == "__main__":
raise SystemExit(main())

288
scripts/security/source-control-migration-inventory.py Normal file
View File

@@ -0,0 +1,288 @@
#!/usr/bin/env python3
"""Read-only Gitea/GitHub 遷移盤點工具。
此工具只用 `git ls-remote` 比對兩端 refs，不 push、不 fetch、不寫入任一 remote。
寫入 evidence 前會先遮蔽 remote URL 內的帳密。
"""
from __future__ import annotations
import argparse
import json
import re
import subprocess
import sys
from dataclasses import dataclass
from pathlib import Path
from urllib.parse import urlsplit, urlunsplit
SHA_RE = re.compile(r"^[0-9a-fA-F]{7,40}$")
@dataclass(frozen=True)
class GitRefs:
heads: dict[str, str]
tags: dict[str, str]
raw_tag_ref_count: int
def run_git(repo: Path, args: list[str]) -> subprocess.CompletedProcess[str]:
return subprocess.run(
["git", *args],
cwd=repo,
check=False,
capture_output=True,
text=True,
)
def require_git(repo: Path, args: list[str]) -> str:
result = run_git(repo, args)
if result.returncode != 0:
stderr = result.stderr.strip() or result.stdout.strip()
raise RuntimeError(f"git {' '.join(args)} failed: {stderr}")
return result.stdout
def redact_url(url: str) -> str:
"""Remove userinfo from common URL formats before storing evidence."""
if "://" in url:
parts = urlsplit(url)
netloc = parts.netloc
if "@" in netloc:
netloc = netloc.split("@", 1)[1]
return urlunsplit((parts.scheme, netloc, parts.path, parts.query, parts.fragment))
# scp-like syntax: user@host:path
if "@" in url and ":" in url.split("@", 1)[1]:
return url.split("@", 1)[1]
return url
def remote_url(repo: Path, remote: str) -> str:
return redact_url(require_git(repo, ["remote", "get-url", remote]).strip())
def repo_slug_from_url(url: str) -> str:
redacted = redact_url(url).removesuffix("/")
if "://" in redacted:
path = urlsplit(redacted).path.strip("/")
elif ":" in redacted:
path = redacted.split(":", 1)[1].strip("/")
else:
path = redacted.strip("/")
return path.removesuffix(".git") or redacted
def parse_ls_remote(output: str, prefix: str) -> dict[str, str]:
refs: dict[str, str] = {}
for line in output.splitlines():
if not line.strip():
continue
try:
sha, ref = line.split(None, 1)
except ValueError:
continue
if not SHA_RE.match(sha) or not ref.startswith(prefix):
continue
name = ref.removeprefix(prefix)
refs[name] = sha
return refs
def refs_for_remote(repo: Path, remote: str) -> GitRefs:
heads_out = require_git(repo, ["ls-remote", "--heads", remote])
tags_out = require_git(repo, ["ls-remote", "--tags", remote])
raw_tags = parse_ls_remote(tags_out, "refs/tags/")
tags = {
name.removesuffix("^{}"): sha
for name, sha in raw_tags.items()
if not name.endswith("^{}")
}
peeled = {
name.removesuffix("^{}"): sha
for name, sha in raw_tags.items()
if name.endswith("^{}")
}
tags.update({name: peeled_sha for name, peeled_sha in peeled.items()})
return GitRefs(
heads=parse_ls_remote(heads_out, "refs/heads/"),
tags=tags,
raw_tag_ref_count=len(raw_tags),
)
def compare_maps(left: dict[str, str], right: dict[str, str]) -> dict[str, object]:
left_names = set(left)
right_names = set(right)
common = sorted(left_names & right_names)
return {
"only_left": sorted(left_names - right_names),
"only_right": sorted(right_names - left_names),
"sha_mismatch": [
{
"name": name,
"left_sha": left[name],
"right_sha": right[name],
}
for name in common
if left[name] != right[name]
],
"matching": [name for name in common if left[name] == right[name]],
}
def build_inventory(repo: Path, gitea_remote: str, github_remote: str) -> dict[str, object]:
gitea = refs_for_remote(repo, gitea_remote)
github = refs_for_remote(repo, github_remote)
gitea_url = remote_url(repo, gitea_remote)
github_url = remote_url(repo, github_remote)
head_diff = compare_maps(gitea.heads, github.heads)
tag_diff = compare_maps(gitea.tags, github.tags)
latest_sha_gitea = gitea.heads.get("main", "")
latest_sha_github = github.heads.get("main", "")
status = "verified"
blocking_reasons: list[str] = []
if head_diff["only_left"] or head_diff["only_right"] or head_diff["sha_mismatch"]:
status = "blocked"
blocking_reasons.append("branches 尚未完全對齊")
if tag_diff["only_left"] or tag_diff["only_right"] or tag_diff["sha_mismatch"]:
status = "blocked"
blocking_reasons.append("tags 尚未完全對齊")
if latest_sha_gitea and latest_sha_github and latest_sha_gitea != latest_sha_github:
status = "blocked"
blocking_reasons.append("main SHA 不一致")
if not latest_sha_gitea or not latest_sha_github:
status = "blocked"
blocking_reasons.append("其中一端缺少 main")
return {
"repo_path": str(repo),
"gitea_remote": gitea_remote,
"github_remote": github_remote,
"gitea_url_redacted": gitea_url,
"github_url_redacted": github_url,
"gitea_repo": repo_slug_from_url(gitea_url),
"github_repo": repo_slug_from_url(github_url),
"branch_count_gitea": len(gitea.heads),
"branch_count_github": len(github.heads),
"tag_count_gitea": len(gitea.tags),
"tag_count_github": len(github.tags),
"raw_tag_ref_count_gitea": gitea.raw_tag_ref_count,
"raw_tag_ref_count_github": github.raw_tag_ref_count,
"latest_sha_gitea": latest_sha_gitea,
"latest_sha_github": latest_sha_github,
"workflows_mapped": False,
"webhooks_mapped": False,
"secrets_inventory_only": True,
"status": status,
"blocking_reason": "；".join(blocking_reasons) if blocking_reasons else "",
"heads": head_diff,
"tags": tag_diff,
}
def event_payload(inventory: dict[str, object], evidence_ref: str | None) -> dict[str, object]:
payload = {
"schema_version": "source_control_migration_event_v1",
"gitea_repo": inventory["gitea_repo"],
"github_repo": inventory["github_repo"],
"branch_count_gitea": inventory["branch_count_gitea"],
"branch_count_github": inventory["branch_count_github"],
"tag_count_gitea": inventory["tag_count_gitea"],
"tag_count_github": inventory["tag_count_github"],
"latest_sha_gitea": inventory["latest_sha_gitea"],
"latest_sha_github": inventory["latest_sha_github"],
"workflows_mapped": inventory["workflows_mapped"],
"webhooks_mapped": inventory["webhooks_mapped"],
"secrets_inventory_only": inventory["secrets_inventory_only"],
"status": inventory["status"],
"blocking_reason": inventory["blocking_reason"],
}
if evidence_ref:
payload["evidence_refs"] = [evidence_ref]
return payload
def write_markdown(inventory: dict[str, object], path: Path) -> None:
heads = inventory["heads"]
tags = inventory["tags"]
assert isinstance(heads, dict)
assert isinstance(tags, dict)
lines = [
"# Source Control 遷移盤點快照",
"",
"| 項目 | 值 |",
"|------|----|",
f"| 狀態 | `{inventory['status']}` |",
f"| Gitea remote | `{inventory['gitea_remote']}` |",
f"| GitHub remote | `{inventory['github_remote']}` |",
f"| Gitea repo | `{inventory['gitea_repo']}` |",
f"| GitHub repo | `{inventory['github_repo']}` |",
f"| Gitea URL | `{inventory['gitea_url_redacted']}` |",
f"| GitHub URL | `{inventory['github_url_redacted']}` |",
f"| Gitea 分支數 | `{inventory['branch_count_gitea']}` |",
f"| GitHub 分支數 | `{inventory['branch_count_github']}` |",
f"| Gitea tags | `{inventory['tag_count_gitea']}` |",
f"| GitHub tags | `{inventory['tag_count_github']}` |",
f"| Gitea main | `{inventory['latest_sha_gitea']}` |",
f"| GitHub main | `{inventory['latest_sha_github']}` |",
f"| 阻塞原因 | {inventory['blocking_reason'] or '無'} |",
"",
"## 分支差異",
"",
f"- 只在 Gitea：`{len(heads['only_left'])}`",
f"- 只在 GitHub：`{len(heads['only_right'])}`",
f"- SHA 不一致：`{len(heads['sha_mismatch'])}`",
f"- SHA 一致：`{len(heads['matching'])}`",
"",
"## Tag 差異",
"",
f"- 只在 Gitea：`{len(tags['only_left'])}`",
f"- 只在 GitHub：`{len(tags['only_right'])}`",
f"- SHA 不一致：`{len(tags['sha_mismatch'])}`",
f"- SHA 一致：`{len(tags['matching'])}`",
"",
"> 注意：本檔由 read-only inventory 工具產生，不包含 remote URL 內的帳密。",
"",
]
path.write_text("\n".join(lines), encoding="utf-8")
def main() -> int:
parser = argparse.ArgumentParser()
parser.add_argument("--repo", default=".")
parser.add_argument("--gitea-remote", default="gitea")
parser.add_argument("--github-remote", default="origin")
parser.add_argument("--output-json")
parser.add_argument("--output-md")
args = parser.parse_args()
repo = Path(args.repo).resolve()
try:
inventory = build_inventory(repo, args.gitea_remote, args.github_remote)
except RuntimeError as exc:
print(str(exc), file=sys.stderr)
return 2
payload = json.dumps(
event_payload(inventory, args.output_md),
ensure_ascii=False,
indent=2,
)
if args.output_json:
Path(args.output_json).write_text(payload + "\n", encoding="utf-8")
else:
print(payload)
if args.output_md:
write_markdown(inventory, Path(args.output_md))
return 0
if __name__ == "__main__":
raise SystemExit(main())