diff --git a/gitea/gitea-terraforming.py b/gitea/gitea-terraforming.py new file mode 100755 index 0000000..2159f37 --- /dev/null +++ b/gitea/gitea-terraforming.py @@ -0,0 +1,274 @@ +#!/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-.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 --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 --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()