buildbot-nix/buildbot_nix/github_projects.py

import http.client
import json
import os
import urllib.request
from pathlib import Path
from tempfile import NamedTemporaryFile
from typing import Any

from twisted.python import log


class HttpResponse:
    def __init__(self, raw: http.client.HTTPResponse) -> None:
        self.raw = raw

    def json(self) -> Any:
        return json.load(self.raw)

    def headers(self) -> http.client.HTTPMessage:
        return self.raw.headers


def http_request(
    url: str,
    method: str = "GET",
    headers: dict[str, str] = {},
    data: dict[str, Any] | None = None,
) -> HttpResponse:
    body = None
    if data:
        body = json.dumps(data).encode("ascii")
    headers = headers.copy()
    headers["User-Agent"] = "buildbot-nix"
    req = urllib.request.Request(url, headers=headers, method=method, data=body)
    try:
        resp = urllib.request.urlopen(req)
    except urllib.request.HTTPError as e:
        resp_body = ""
        try:
            resp_body = e.fp.read().decode("utf-8", "replace")
        except Exception:
            pass
        raise Exception(
            f"Request for {method} {url} failed with {e.code} {e.reason}: {resp_body}"
        ) from e
    return HttpResponse(resp)


def paginated_github_request(url: str, token: str) -> list[dict[str, Any]]:
    next_url: str | None = url
    items = []
    while next_url:
        try:
            res = http_request(
                next_url,
                headers={"Authorization": f"Bearer {token}"},
            )
        except OSError as e:
            raise Exception(f"failed to fetch {next_url}: {e}") from e
        next_url = None
        link = res.headers()["Link"]
        if link is not None:
            links = link.split(", ")
            for link in links:  # pagination
                link_parts = link.split(";")
                if link_parts[1].strip() == 'rel="next"':
                    next_url = link_parts[0][1:-1]
        items += res.json()
    return items


class GithubProject:
    def __init__(self, data: dict[str, Any]) -> None:
        self.data = data

    @property
    def repo(self) -> str:
        return self.data["name"]

    @property
    def owner(self) -> str:
        return self.data["owner"]["login"]

    @property
    def name(self) -> str:
        return self.data["full_name"]

    @property
    def url(self) -> str:
        return self.data["html_url"]

    @property
    def id(self) -> str:
        n = self.data["full_name"]
        return n.replace("/", "-")

    @property
    def default_branch(self) -> str:
        return self.data["default_branch"]

    @property
    def topics(self) -> list[str]:
        return self.data["topics"]


def create_project_hook(
    owner: str, repo: str, token: str, webhook_url: str, webhook_secret: str
) -> None:
    hooks = paginated_github_request(
        f"https://api.github.com/repos/{owner}/{repo}/hooks?per_page=100", token
    )
    config = dict(
        url=webhook_url, content_type="json", insecure_ssl="0", secret=webhook_secret
    )
    data = dict(name="web", active=True, events=["push", "pull_request"], config=config)
    headers = {
        "Authorization": f"Bearer {token}",
        "Accept": "application/vnd.github+json",
        "Content-Type": "application/json",
        "X-GitHub-Api-Version": "2022-11-28",
    }
    for hook in hooks:
        if hook["config"]["url"] == webhook_url:
            log.msg(f"hook for {owner}/{repo} already exists")
            return

    http_request(
        f"https://api.github.com/repos/{owner}/{repo}/hooks",
        method="POST",
        headers=headers,
        data=data,
    )


def refresh_projects(github_token: str, repo_cache_file: Path) -> None:
    repos = []

    for repo in paginated_github_request(
            "https://api.github.com/user/repos?per_page=100",
            github_token,
    ):
        if not repo["permissions"]["admin"]:
            name = repo['full_name']
            log.msg(f"skipping {name} because we do not have admin privileges, needed for hook management")
        else:
            repos.append(repo)

    with NamedTemporaryFile("w", delete=False, dir=repo_cache_file.parent) as f:
        try:
            f.write(json.dumps(repos))
            f.flush()
            os.rename(f.name, repo_cache_file)
        except OSError:
            os.unlink(f.name)
            raise


def load_projects(github_token: str, repo_cache_file: Path) -> list[GithubProject]:
    if not repo_cache_file.exists():
        log.msg("fetching github repositories")
        refresh_projects(github_token, repo_cache_file)
    repos: list[dict[str, Any]] = json.loads(repo_cache_file.read_text())
    return [GithubProject(repo) for repo in repos]