add buildbot-effects

This is an implementation of hercules-ci-effects in python.

Author: Mic92

bin/buildbot-effects

@@ -0,0 +1,9 @@
+#!/usr/bin/env python
+import sys
+from pathlib import Path
+sys.path.append(str(Path(__file__).parent.parent))
+
+from hercules_effects.cli import main
+
+if __name__ == '__main__':
+    main()

buildbot_effects/__init__.py

@@ -0,0 +1,233 @@
+import json
+import os
+import shlex
+import shutil
+import subprocess
+import sys
+from collections.abc import Iterator
+from contextlib import contextmanager
+from pathlib import Path
+from tempfile import NamedTemporaryFile
+from typing import IO, Any
+
+from .options import EffectsOptions
+
+
+def run(
+    cmd: list[str],
+    stdin: int | IO[str] | None = None,
+    stdout: int | IO[str] | None = None,
+    stderr: int | IO[str] | None = None,
+    verbose: bool = True,
+) -> subprocess.CompletedProcess[str]:
+    if verbose:
+        print("$", shlex.join(cmd), file=sys.stderr)
+    return subprocess.run(
+        cmd,
+        check=True,
+        text=True,
+        stdin=stdin,
+        stdout=stdout,
+        stderr=stderr,
+    )
+
+
+def git_command(args: list[str], path: Path) -> str:
+    cmd = ["git", "-C", str(path), *args]
+    proc = run(cmd, stdout=subprocess.PIPE)
+    return proc.stdout.strip()
+
+
+def get_git_rev(path: Path) -> str:
+    return git_command(["rev-parse", "--verify", "HEAD"], path)
+
+
+def get_git_branch(path: Path) -> str:
+    return git_command(["rev-parse", "--abbrev-ref", "HEAD"], path)
+
+
+def get_git_remote_url(path: Path) -> str | None:
+    try:
+        return git_command(["remote", "get-url", "origin"], path)
+    except subprocess.CalledProcessError:
+        return None
+
+
+def git_get_tag(path: Path, rev: str) -> str | None:
+    tags = git_command(["tag", "--points-at", rev], path)
+    if tags:
+        return tags.splitlines()[1]
+    return None
+
+
+def effects_args(opts: EffectsOptions) -> dict[str, Any]:
+    rev = opts.rev or get_git_rev(opts.path)
+    short_rev = rev[:7]
+    branch = opts.branch or get_git_branch(opts.path)
+    repo = opts.repo or opts.path.name
+    tag = opts.tag or git_get_tag(opts.path, rev)
+    url = opts.url or get_git_remote_url(opts.path)
+    primary_repo = dict(
+        name=repo,
+        branch=branch,
+        # TODO: support ref
+        ref=None,
+        tag=tag,
+        rev=rev,
+        shortRev=short_rev,
+        remoteHttpUrl=url,
+    )
+    return {
+        "primaryRepo": primary_repo,
+        **primary_repo,
+    }
+
+
+def nix_command(*args: str) -> list[str]:
+    return ["nix", "--extra-experimental-features", "nix-command flakes", *args]
+
+
+def effect_function(opts: EffectsOptions) -> str:
+    args = effects_args(opts)
+    rev = args["rev"]
+    escaped_args = json.dumps(json.dumps(args))
+    url = json.dumps(f"git+file://{opts.path}?rev={rev}#")
+    return f"""(((builtins.getFlake {url}).outputs.herculesCI (builtins.fromJSON {escaped_args})).onPush.default.outputs.hci-effects)"""
+
+
+def list_effects(opts: EffectsOptions) -> list[str]:
+    cmd = nix_command(
+        "eval",
+        "--json",
+        "--expr",
+        f"builtins.attrNames {effect_function(opts)}",
+    )
+    proc = run(cmd, stdout=subprocess.PIPE)
+    return json.loads(proc.stdout)
+
+
+def instantiate_effects(opts: EffectsOptions) -> str:
+    cmd = [
+        "nix-instantiate",
+        "--expr",
+        f"{effect_function(opts)}.deploy.run",
+    ]
+    proc = run(cmd, stdout=subprocess.PIPE)
+    return proc.stdout.rstrip()
+
+
+def parse_derivation(path: str) -> dict[str, Any]:
+    cmd = [
+        "nix",
+        "--extra-experimental-features",
+        "nix-command flakes",
+        "derivation",
+        "show",
+        f"{path}^*",
+    ]
+    proc = run(cmd, stdout=subprocess.PIPE)
+    return json.loads(proc.stdout)
+
+
+def env_args(env: dict[str, str]) -> list[str]:
+    result = []
+    for k, v in env.items():
+        result.append("--setenv")
+        result.append(f"{k}")
+        result.append(f"{v}")
+    return result
+
+
+@contextmanager
+def pipe() -> Iterator[tuple[IO[str], IO[str]]]:
+    r, w = os.pipe()
+    r_file = os.fdopen(r, "r")
+    w_file = os.fdopen(w, "w")
+    try:
+        yield r_file, w_file
+    finally:
+        r_file.close()
+        w_file.close()
+
+
+def run_effects(
+    drv_path: str, drv: dict[str, Any], secrets: dict[str, Any] = {}
+) -> None:
+    builder = drv["builder"]
+    args = drv["args"]
+    sandboxed_cmd = [
+        builder,
+        *args,
+    ]
+    env = {}
+    env["IN_HERCULES_CI_EFFECT"] = "true"
+    env["HERCULES_CI_SECRETS_JSON"] = "/run/secrets.json"
+    env["NIX_BUILD_TOP"] = "/build"
+    bwrap = shutil.which("bwrap")
+    if bwrap is None:
+        raise Exception("bwrap not found")
+
+    bubblewrap_cmd = [
+        "nix",
+        "develop",
+        "-i",
+        f"{drv_path}^*",
+        "-c",
+        bwrap,
+        "--unshare-all",
+        "--share-net",
+        "--new-session",
+        "--die-with-parent",
+        "--dir",
+        "/build",
+        "--chdir",
+        "/build",
+        "--tmpfs",
+        "/tmp",
+        "--tmpfs",
+        "/build",
+        "--proc",
+        "/proc",
+        "--dev",
+        "/dev",
+        "--ro-bind",
+        "/etc/resolv.conf",
+        "/etc/resolv.conf",
+        "--ro-bind",
+        "/etc/hosts",
+        "/etc/hosts",
+        "--ro-bind",
+        "/nix/store",
+        "/nix/store",
+    ]
+
+    with NamedTemporaryFile() as tmp:
+        secrets = secrets.copy()
+        secrets["hercules-ci"] = {"data": {"token": "dummy"}}
+        tmp.write(json.dumps(secrets).encode())
+        bubblewrap_cmd.extend(
+            [
+                "--ro-bind",
+                tmp.name,
+                "/run/secrets.json",
+            ]
+        )
+        bubblewrap_cmd.extend(env_args(env))
+        bubblewrap_cmd.append("--")
+        bubblewrap_cmd.extend(sandboxed_cmd)
+        with pipe() as (r_file, w_file):
+            print("$", shlex.join(bubblewrap_cmd), file=sys.stderr)
+            proc = subprocess.Popen(
+                bubblewrap_cmd,
+                text=True,
+                stdin=subprocess.DEVNULL,
+                stdout=w_file,
+                stderr=w_file,
+            )
+            w_file.close()
+            with proc:
+                for line in r_file:
+                    print(line, end="")
+                proc.wait()
+                if proc.returncode != 0:
+                    raise Exception(f"command failed with exit code {proc.returncode}")

buildbot_effects/cli.py

@@ -0,0 +1,88 @@
+import argparse
+from collections.abc import Callable
+from pathlib import Path
+import json
+
+from .options import EffectsOptions
+from . import list_effects, instantiate_effects, parse_derivation, run_effects
+
+
+def list_command(options: EffectsOptions) -> None:
+    print(list_effects(options))
+
+
+def run_command(options: EffectsOptions) -> None:
+    drv_path = instantiate_effects(options)
+    drvs = parse_derivation(drv_path)
+    drv = next(iter(drvs.values()))
+
+    if options.secrets:
+        secrets = json.loads(options.secrets.read_text())
+    else:
+        secrets = {}
+    run_effects(drv_path, drv, secrets=secrets)
+
+
+def run_all_command(options: EffectsOptions) -> None:
+    print("TODO")
+
+
+def parse_args() -> tuple[Callable[[EffectsOptions], None], EffectsOptions]:
+    parser = argparse.ArgumentParser(description="Run effects from a hercules-ci flake")
+    parser.add_argument(
+        "--secrets",
+        type=Path,
+        help="Path to a json file with secrets",
+    )
+    parser.add_argument(
+        "--rev",
+        type=str,
+        help="Git revision to use",
+    )
+    parser.add_argument(
+        "--branch",
+        type=str,
+        help="Git branch to use",
+    )
+    parser.add_argument(
+        "--repo",
+        type=str,
+        help="Git repo to prepend to be",
+    )
+    parser.add_argument(
+        "--path",
+        type=str,
+        help="Path to the repository",
+    )
+    subparser = parser.add_subparsers(
+        dest="command",
+        required=True,
+        help="Command to run",
+    )
+    list_parser = subparser.add_parser(
+        "list",
+        help="List available effects",
+    )
+    list_parser.set_defaults(command=list_command)
+    run_parser = subparser.add_parser(
+        "run",
+        help="Run an effect",
+    )
+    run_parser.set_defaults(command=run_command)
+    run_parser.add_argument(
+        "effect",
+        help="Effect to run",
+    )
+    run_all_parser = subparser.add_parser(
+        "run-all",
+        help="Run all effects",
+    )
+    run_all_parser.set_defaults(command=run_all_command)
+
+    args = parser.parse_args()
+    return args.command, EffectsOptions(secrets=args.secrets)
+
+
+def main() -> None:
+    command, options = parse_args()
+    command(options)

buildbot_effects/options.py

@@ -0,0 +1,13 @@
+from dataclasses import dataclass, field
+from pathlib import Path
+
+
+@dataclass
+class EffectsOptions:
+    secrets: Path | None = None
+    path: Path = field(default_factory=lambda: Path.cwd())
+    repo: str | None = ""
+    rev: str | None = None
+    branch: str | None = None
+    url: str | None = None
+    tag: str | None = None

pyproject.toml

@@ -21,9 +21,13 @@ classifiers = [
     "Programming Language :: Python"
 ]
 version = "0.0.1"
+scripts = { buildbot-effects = "hercules_effects.cli:main" }
 
 [tool.setuptools]
-packages = ["buildbot_nix"]
+packages = [
+  "buildbot_nix",
+  "buildbot_effects"
+]
 
 [tool.ruff]
 target-version = "py311"