275 lines
8.8 KiB
Python
Executable File
275 lines
8.8 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""
|
|
gitea-terraforming: Reverse Terraform for Gitea (OpenTofu compatible)
|
|
|
|
Generates Terraform HCL for:
|
|
- Users
|
|
- Organizations
|
|
- Repositories (user & org)
|
|
- Branch protections
|
|
|
|
Output files are automatically split per resource type:
|
|
gitea-<resource-type>.tf
|
|
|
|
Supports import generation:
|
|
- Modern import blocks (--modern-import-block)
|
|
- Shell script terraform import
|
|
|
|
Usage example:
|
|
python gitea_terraforming.py --api https://gitea.example.com --token <ADMIN_TOKEN> --out-dir ./gitea_tf
|
|
"""
|
|
|
|
import argparse, os, sys, time, json, re
|
|
from typing import Any, Dict, List, Optional
|
|
from datetime import datetime
|
|
import requests
|
|
|
|
def slugify(s: str) -> str:
|
|
s = re.sub(r'[^0-9a-zA-Z_-]', '_', s)
|
|
return re.sub('_+', '_', s).strip('_').lower()
|
|
|
|
class GiteaClient:
|
|
def __init__(self, api_base: str, token: str, verify: bool = True):
|
|
self.base = api_base.rstrip('/')
|
|
self.s = requests.Session()
|
|
self.s.headers.update({
|
|
'Authorization': f'token {token}',
|
|
'Accept': 'application/json',
|
|
'User-Agent': 'gitea-terraforming/0.1'
|
|
})
|
|
self.verify = verify
|
|
|
|
def _get(self, path: str, params: Optional[dict] = None):
|
|
url = f"{self.base}{path}"
|
|
out = []
|
|
page = 1
|
|
while True:
|
|
qp = params.copy() if params else {}
|
|
qp.update({'page': page, 'limit': 100})
|
|
resp = self.s.get(url, params=qp, verify=self.verify, timeout=30)
|
|
if resp.status_code == 404:
|
|
return []
|
|
if resp.status_code == 429:
|
|
retry = int(resp.headers.get('Retry-After', '5'))
|
|
time.sleep(retry)
|
|
continue
|
|
resp.raise_for_status()
|
|
data = resp.json()
|
|
if isinstance(data, list):
|
|
out.extend(data)
|
|
if len(data) < 100:
|
|
break
|
|
page += 1
|
|
else:
|
|
return data
|
|
return out
|
|
|
|
def list_orgs(self) -> List[dict]:
|
|
return self._get("/api/v1/orgs")
|
|
|
|
def list_users(self) -> List[dict]:
|
|
return self._get("/api/v1/admin/users")
|
|
|
|
def list_user_repos(self, user: str) -> List[dict]:
|
|
return self._get(f"/api/v1/users/{user}/repos")
|
|
|
|
def list_org_repos(self, org: str) -> List[dict]:
|
|
return self._get(f"/api/v1/orgs/{org}/repos")
|
|
|
|
def list_branch_protections(self, owner: str, repo: str) -> List[dict]:
|
|
return self._get(f"/api/v1/repos/{owner}/{repo}/branch_protections")
|
|
|
|
def hcl_block(resource_type: str, name: str, attrs: dict, comment: Optional[str] = None) -> str:
|
|
lines = []
|
|
if comment:
|
|
lines.append(f"# {comment}")
|
|
lines.append(f'resource "{resource_type}" "{name}" ' + "{")
|
|
for k, v in attrs.items():
|
|
if v is None:
|
|
continue
|
|
if isinstance(v, bool):
|
|
lines.append(f" {k} = {str(v).lower()}")
|
|
elif isinstance(v, (int, float)):
|
|
lines.append(f" {k} = {v}")
|
|
elif isinstance(v, str):
|
|
safe = v.replace('"', '\\"')
|
|
lines.append(f' {k} = "{safe}"')
|
|
elif isinstance(v, list):
|
|
joined = ", ".join([f'"{x}"' for x in v])
|
|
lines.append(f" {k} = [{joined}]")
|
|
else:
|
|
lines.append(f' # {k} = {json.dumps(v)}')
|
|
lines.append("}\n")
|
|
return "\n".join(lines)
|
|
|
|
def modern_import_block(to: str, ident: str) -> str:
|
|
return f'import {{\n to = {to}\n id = "{ident}"\n}}\n'
|
|
|
|
def generate(api: str, token: str, out_dir: str, modern: bool = False, dry: bool = False):
|
|
client = GiteaClient(api, token)
|
|
os.makedirs(out_dir, exist_ok=True)
|
|
imports = []
|
|
files: dict = {}
|
|
|
|
orgs = client.list_orgs()
|
|
org_buf = []
|
|
for o in orgs:
|
|
uname = o.get("username") or o.get("user_name") or o.get("name")
|
|
rname = f"org_{slugify(uname)}"
|
|
attrs = {"name": uname}
|
|
org_buf.append(hcl_block("gitea_organization", rname, attrs, comment=f"source id={o.get('id')}"))
|
|
imports.append((f"gitea_organization.{rname}", uname))
|
|
if org_buf:
|
|
files["orgs"] = "\n".join(org_buf)
|
|
|
|
users = []
|
|
try:
|
|
users = client.list_users()
|
|
except Exception as e:
|
|
print(f"Warning: cannot list users: {e}", file=sys.stderr)
|
|
user_buf = []
|
|
for u in users:
|
|
uname = u.get("login") or u.get("username")
|
|
if not uname:
|
|
continue
|
|
rname = f"user_{slugify(uname)}"
|
|
attrs = {
|
|
"username": uname,
|
|
"email": u.get("email"),
|
|
"full_name": u.get("full_name"),
|
|
"is_admin": u.get("is_admin")
|
|
}
|
|
user_buf.append(hcl_block("gitea_user", rname, attrs, comment=f"source id={u.get('id')}"))
|
|
imports.append((f"gitea_user.{rname}", uname))
|
|
if user_buf:
|
|
files["users"] = "\n".join(user_buf)
|
|
|
|
repo_buf = []
|
|
for u in users:
|
|
uname = u.get("login") or u.get("username")
|
|
if not uname:
|
|
continue
|
|
try:
|
|
repos = client.list_user_repos(uname)
|
|
except Exception:
|
|
repos = []
|
|
for r in repos:
|
|
rname = f"repo_{slugify(uname)}_{slugify(r['name'])}"
|
|
attrs = {
|
|
"owner": uname,
|
|
"name": r["name"],
|
|
"private": r.get("private", False),
|
|
"description": r.get("description")
|
|
}
|
|
repo_buf.append(hcl_block("gitea_repository", rname, attrs, comment=f"source id={r.get('id')}"))
|
|
imports.append((f"gitea_repository.{rname}", f"{uname}/{r['name']}"))
|
|
for o in orgs:
|
|
uname = o.get("username") or o.get("user_name") or o.get("name")
|
|
if not uname:
|
|
continue
|
|
try:
|
|
repos = client.list_org_repos(uname)
|
|
except Exception:
|
|
repos = []
|
|
for r in repos:
|
|
rname = f"repo_{slugify(uname)}_{slugify(r['name'])}"
|
|
attrs = {
|
|
"owner": uname,
|
|
"name": r["name"],
|
|
"private": r.get("private", False),
|
|
"description": r.get("description")
|
|
}
|
|
repo_buf.append(hcl_block("gitea_repository", rname, attrs, comment=f"source id={r.get('id')}"))
|
|
imports.append((f"gitea_repository.{rname}", f"{uname}/{r['name']}"))
|
|
if repo_buf:
|
|
files["repos"] = "\n".join(repo_buf)
|
|
|
|
bp_buf = []
|
|
for to, ident in imports:
|
|
if not to.startswith("gitea_repository."):
|
|
continue
|
|
owner, repo = ident.split("/", 1)
|
|
try:
|
|
bps = client.list_branch_protections(owner, repo)
|
|
except Exception:
|
|
bps = []
|
|
for bp in bps:
|
|
branch_name = bp.get("branch_name") or bp.get("branch") or bp.get("name")
|
|
if branch_name is None:
|
|
continue
|
|
bn = f"branch_protection_{slugify(owner)}_{slugify(repo)}_{slugify(branch_name)}"
|
|
attrs = {
|
|
"repository": repo,
|
|
"owner": owner,
|
|
"branch": branch_name,
|
|
"enable_status_check": bp.get("enable_status_check", False),
|
|
"required_approvals": bp.get("required_approvals", 0),
|
|
"enable_merge_whitelist": bp.get("enable_merge_whitelist", False),
|
|
}
|
|
bp_buf.append(hcl_block("gitea_branch_protection", bn, attrs, comment=f"protect branch {branch_name}"))
|
|
imports.append((f"gitea_branch_protection.{bn}", f"{owner}/{repo}/{branch_name}"))
|
|
if bp_buf:
|
|
files["branches"] = "\n".join(bp_buf)
|
|
|
|
for rtype, content in files.items():
|
|
fname = f"gitea-{rtype}.tf"
|
|
fpath = os.path.join(out_dir, fname)
|
|
if dry:
|
|
print(f"--- {fpath} ---")
|
|
print(content)
|
|
else:
|
|
with open(fpath, "w", encoding="utf-8") as f:
|
|
f.write(f"// Generated by gitea-terraforming (OpenTofu compatible)\n\n")
|
|
f.write(content)
|
|
print(f"Wrote {fname}", file=sys.stderr)
|
|
|
|
if modern:
|
|
imps = "\n".join([modern_import_block(to, ident) for to, ident in imports])
|
|
imppath = os.path.join(out_dir, "imports.tf")
|
|
if dry:
|
|
print("--- imports.tf ---")
|
|
print(imps)
|
|
else:
|
|
with open(imppath, "w", encoding="utf-8") as f:
|
|
f.write("// Import blocks\n\n")
|
|
f.write(imps)
|
|
print(f"Wrote imports.tf", file=sys.stderr)
|
|
else:
|
|
lines = ["#!/usr/bin/env bash", "set -euo pipefail"]
|
|
lines += [f'terraform import {to} "{ident}"' for to, ident in imports]
|
|
imppath = os.path.join(out_dir, "terraform_imports.sh")
|
|
if dry:
|
|
print("--- terraform_imports.sh ---")
|
|
print("\n".join(lines))
|
|
else:
|
|
with open(imppath, "w", encoding="utf-8") as f:
|
|
f.write("\n".join(lines) + "\n")
|
|
os.chmod(imppath, 0o755)
|
|
print(f"Wrote terraform_imports.sh", file=sys.stderr)
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(
|
|
description="gitea-terraforming: Reverse Terraform for Gitea (OpenTofu compatible)\n\n"
|
|
"Generates Terraform HCL for users, organizations, repositories, and branch protections.\n"
|
|
"Output files are split per resource type and timestamped.\n\n"
|
|
"Example usage:\n"
|
|
" python gitea_terraforming.py --api https://gitea.example.com --token <ADMIN_TOKEN> --out-dir ./gitea_tf\n",
|
|
formatter_class=argparse.RawTextHelpFormatter
|
|
)
|
|
parser.add_argument("--api", required=True, help="Gitea API URL (e.g., https://gitea.example.com)")
|
|
parser.add_argument("--token", default=os.environ.get("GITEA_TOKEN"),
|
|
help="Gitea admin token (or set GITEA_TOKEN environment variable)")
|
|
parser.add_argument("--out-dir", default="./gitea_tf", help="Directory to write Terraform files")
|
|
parser.add_argument("--modern-import-block", action="store_true",
|
|
help="Generate modern OpenTofu import blocks (imports.tf) instead of shell script")
|
|
parser.add_argument("--dry-run", action="store_true", help="Print output instead of writing files")
|
|
args = parser.parse_args()
|
|
|
|
if not args.token:
|
|
parser.error("Missing Gitea token: provide --token or set GITEA_TOKEN environment variable")
|
|
|
|
generate(args.api, args.token, args.out_dir, modern=args.modern_import_block, dry=args.dry_run)
|
|
|
|
if __name__ == "__main__":
|
|
main()
|