guanghu-ice-heart/server-tools/lake-lamp-authz/repo-push-broker.js

#!/usr/bin/env node
"use strict";

// The receiver is deliberately separate from the public authorization process.
// It owns the private Forgejo transport configuration and accepts only a bundle
// already bound to a workorder/session by lake-lamp-authz.
const fs = require("node:fs");
const os = require("node:os");
const path = require("node:path");
const crypto = require("node:crypto");
const { fileURLToPath } = require("node:url");
const { execFile } = require("node:child_process");

function loadRegistry(file) {
  const parsed = JSON.parse(fs.readFileSync(file, "utf8"));
  if (!parsed || !parsed.repos || typeof parsed.repos !== "object") throw new Error("invalid_repo_push_registry");
  return parsed.repos;
}

async function receiveBundle(request, options = {}) {
  const registry = options.registry || loadRegistry(options.registryFile || process.env.LAKE_LAMP_REPO_PUSH_REGISTRY || "/etc/guanghu/lake-lamp/repo-push-registry.json");
  const run = options.run || runGit;
  const repo = String(request.repo || "").toLowerCase();
  const branch = String(request.branch || "main");
  const expectedHead = String(request.expected_head || "").toLowerCase();
  const entry = registry[repo];
  if (!entry) return blocked("repo_not_allowlisted");
  const allowedBranches = Array.isArray(entry.branches)
    ? entry.branches.map(String)
    : [String(entry.branch || "main")];
  if (!/^[a-z0-9][a-z0-9._/-]{0,199}$/.test(branch) || !allowedBranches.includes(branch)) return blocked("branch_not_allowlisted");
  if (!/^[0-9a-f]{40}$/.test(expectedHead)) return blocked("expected_head_required");
  const bundlePath = checkedBundlePath(request.bundle_path, options.uploadDir || process.env.LAKE_LAMP_REPO_UPLOAD_DIR || "/var/lib/guanghu/repo-push-uploads");
  if (!bundlePath) return blocked("bundle_path_invalid");

  const temporary = fs.mkdtempSync(path.join(options.tempDir || os.tmpdir(), "lake-lamp-receive-"));
  const bare = path.join(temporary, "quarantine.git");
  try {
    // Clone the registered repository into quarantine instead of initializing an
    // empty repository and fetching refs. A registered source may itself be
    // shallow; cloning preserves its shallow boundary metadata, while a ref-only
    // fetch can leave the prerequisite commit present but its parents unreadable.
    await run(["clone", "--bare", "--no-local", entry.remote, bare]);
    await run(["-C", bare, "bundle", "verify", bundlePath]);
    await run(["-C", bare, "fetch", bundlePath, `refs/heads/${branch}:refs/heads/incoming`]);
    const incoming = (await run(["-C", bare, "rev-parse", "refs/heads/incoming"])).stdout.trim().toLowerCase();
    if (!/^[0-9a-f]{40}$/.test(incoming)) return blocked("bundle_branch_missing");
    const remoteResult = await run(
      ["ls-remote", "--exit-code", "--heads", entry.remote, `refs/heads/${branch}`],
      { allowFailure: true },
    );
    const remoteHead = parseRemoteHead(remoteResult.stdout);
    const expectedMissing = expectedHead === "0".repeat(40);
    if ((expectedMissing && remoteHead) || (!expectedMissing && remoteHead !== expectedHead)) {
      return blocked("expected_head_mismatch", { expected_head: expectedHead, actual_head: remoteHead || "" });
    }
    if (!expectedMissing) {
      const ancestry = await run(["-C", bare, "merge-base", "--is-ancestor", expectedHead, incoming], { allowFailure: true });
      if (!ancestry.ok) return blocked("non_fast_forward_bundle", { expected_head: expectedHead, incoming_commit: incoming });
    }
    const targetPath = localRepositoryPath(entry.remote);
    if (!targetPath) return blocked("repository_receive_path_invalid");
    const transferRef = `refs/guanghu-router/${crypto.randomUUID()}`;
    try {
      await run([
        "-C",
        targetPath,
        "fetch",
        "--no-tags",
        bare,
        `refs/heads/incoming:${transferRef}`,
      ]);
      const transferred = (await run(["-C", targetPath, "rev-parse", transferRef])).stdout.trim().toLowerCase();
      if (transferred !== incoming) return blocked("repository_transfer_mismatch");
      await run([
        "-C",
        targetPath,
        "update-ref",
        `refs/heads/${branch}`,
        incoming,
        expectedMissing ? "0".repeat(40) : expectedHead,
      ]);
    } finally {
      await run(["-C", targetPath, "update-ref", "-d", transferRef], { allowFailure: true });
    }
    return { ok: true, repo, branch, commit_sha: incoming, verification_url: String(entry.verification_url || ""), diagnostic_code: "repo_push_succeeded", deployment: { state: "not_requested", next_step: "仅当冰朔与语言层确认该提交需要部署时，再创建绑定此 SHA 的部署工单并显式派发。" } };
  } catch (error) {
    return blocked("repo_push_receiver_failed", { evidence: String(error && error.message || "receiver failure").slice(0, 600) });
  } finally {
    fs.rmSync(temporary, { recursive: true, force: true });
  }
}

function resolveRepository(remoteUrl, registry) {
  const requested = normalizedSourceUrl(remoteUrl);
  if (!requested) return blocked("local_repository_remote_missing");
  for (const [repo, entry] of Object.entries(registry || {})) {
    const sourceUrls = Array.isArray(entry && entry.source_urls)
      ? entry.source_urls
      : [];
    if (sourceUrls.some(value => normalizedSourceUrl(value) === requested)) {
      return {
        ok: true,
        diagnostic_code: "repository_registered",
        repo,
        branches: Array.isArray(entry.branches)
          ? entry.branches.map(String)
          : [String(entry.branch || "main")],
      };
    }
  }
  return blocked("repository_not_registered");
}

function normalizedSourceUrl(value) {
  return String(value || "").trim().replace(/\/+$/, "").toLowerCase();
}

function localRepositoryPath(remote) {
  try {
    const value = String(remote || "");
    const resolved = value.startsWith("file://") ? fileURLToPath(value) : value;
    return path.isAbsolute(resolved) ? resolved : "";
  } catch {
    return "";
  }
}

function checkedBundlePath(value, uploadDir) {
  try {
    const root = fs.realpathSync(uploadDir);
    const candidate = fs.realpathSync(String(value || ""));
    return candidate.startsWith(`${root}${path.sep}`) && fs.statSync(candidate).isFile() ? candidate : "";
  } catch { return ""; }
}
function parseRemoteHead(output) {
  const match = String(output || "").match(/^([0-9a-f]{40})\s+refs\/heads\//m);
  return match ? match[1].toLowerCase() : "";
}
function blocked(diagnostic_code, extra = {}) { return { ok: false, diagnostic_code, ...extra }; }
function runGit(args, options = {}) {
  return new Promise((resolve, reject) => execFile("git", args, { timeout: 120000, maxBuffer: 1024 * 1024 }, (error, stdout, stderr) => {
    const result = { ok: !error, stdout: String(stdout || ""), stderr: String(stderr || "") };
    if (error && !options.allowFailure) return reject(new Error(result.stderr || error.message));
    resolve(result);
  }));
}

module.exports = {
  loadRegistry,
  receiveBundle,
  checkedBundlePath,
  parseRemoteHead,
  resolveRepository,
  localRepositoryPath,
};