安全触发私有 Jenkins，使用 GitHub Webhooks、API Gateway（公共）和 Lambda（私有）

Source: Dev.to

架构背后的理念

核心思路很简单：

1. API Gateway 是唯一面向公网的组件。
2. Lambda 和 Jenkins 保持在 VPC 内部私有。
3. 在触发 Jenkins 之前，所有内容都会经过验证。

GitHub (Public)
   |
   | HTTPS Webhook
   v
API Gateway (Public REST API)
   |
   v
AWS Lambda (Private – inside VPC)
   |
   v
Jenkins (Private – EC2 / VPC)

结果

• Jenkins 永不直接接触互联网流量。
• Lambda 成为安全闸门。
• 只有经过验证的 GitHub 事件才会触发构建。

我们使用的组件

• API Gateway（REST API） – 公共 webhook 端点。
• AWS Lambda（在 VPC 内）– 验证负载并转发请求。
• Private Jenkins – 在同一 VPC 中运行。
• GitHub HMAC 签名验证 – 确保负载完整性。
• 不使用花哨的服务——只是在恰当位置使用恰当的组件。

步骤‑逐步指南

1️⃣ 设置 Jenkins（私有）

• 在 VPC 内的 EC2 实例（或 VM）上运行 Jenkins。
• 将其放置在 私有子网 中，不分配公网 IP。
• 安全组
  • 入方向：仅允许来自 Lambda 安全组的流量。
  • 出方向：如果 Jenkins 需要访问外部，允许 HTTPS。

此时 Jenkins 完全与互联网隔离。

2️⃣ 创建 Lambda 函数（私有）

• 将函数关联到 私有子网。
• 使用 安全组，允许出站 HTTPS（用于访问 Jenkins）。

3️⃣ 创建 API Gateway（公共入口）

⚠️ 使用 REST API，而不是 HTTP API。
REST API 能够完整控制请求验证和头部。

1. API Gateway → 创建 API → 选择 REST API。
2. 创建资源（例如 /）。
3. 添加 POST 方法。
4. 该 API 将是唯一面向互联网暴露的入口。

4️⃣ 将 API Gateway 连接到 Lambda

• 在 POST 方法的 Integration Request 中：
  • 集成类型：Lambda Function
  • 选择你创建的 Lambda。
• 部署 API（例如，部署到 dev 阶段）。

启用 Lambda 代理集成 —— 它会原样转发头部、正文以及原始请求负载。

5️⃣ 方法请求配置（关键点）

1. 打开 API Gateway → Resources → POST → Method Request。
2. 设置：

为什么重要
若未启用请求验证器，API Gateway 可能会丢弃诸如 X‑Hub‑Signature‑256 的头部。这样 Lambda 中的签名验证就会始终失败。

3. 更改设置后 重新部署 API。

6️⃣ 配置 GitHub Webhook

在你的 GitHub 仓库中：

Settings → Webhooks → Add webhook

GitHub 现在会发送：

• 原始 JSON 负载。
• X‑Hub‑Signature‑256 头部。

Lambda 实际做了什么

Lambda 函数充当 GitHub 与 Jenkins 之间的安全网关。它会：

1. 读取 原始 webhook 负载。
2. 验证 GitHub 的 HMAC 签名。
3. 检查 分支（或其他条件）。
4. 私密触发 Jenkins 任务。

Lambda 代码（生产就绪）

import json
import hmac
import hashlib
import base64
import urllib.parse
import requests
from requests.auth import HTTPBasicAuth

# ----------------------------------------------------------------------
# Configuration
# ----------------------------------------------------------------------
GITHUB_SECRET = b"demo-github-secret"

JENKINS_URL   = "https://jenkins.example.com"
JENKINS_USER  = "demo-user"
JENKINS_TOKEN = "demo-api-token"

JOB_URL = f"{JENKINS_URL}/job/demo-pipeline/buildWithParameters"
# ----------------------------------------------------------------------

def extract_payload(event):
    """
    Preserve the exact raw body GitHub signed.
    """
    # Normalise header keys to lower‑case
    headers = {k.lower(): v for k, v in (event.get("headers") or {}).items()}
    body = event.get("body", "")

    # GitHub may send the payload base64‑encoded (API Gateway default)
    if event.get("isBase64Encoded"):
        raw_body = base64.b64decode(body)
    else:
        raw_body = body.encode("utf-8")

    # Some GitHub integrations wrap the JSON in a `payload=` form‑urlencoded string
    if raw_body.startswith(b"payload="):
        raw_body = urllib.parse.unquote_to_bytes(
            raw_body.decode().replace("payload=", "", 1)
        )
    return raw_body, headers

def verify_signature(raw_body, headers):
    """
    Verify the HMAC SHA‑256 signature sent by GitHub.
    """
    signature_header = headers.get("x-hub-signature-256")
    if not signature_header:
        raise ValueError("Missing X-Hub-Signature-256 header")

    # Header format: sha256=abcdef...
    try:
        _, received_sig = signature_header.split("=")
    except ValueError:
        raise ValueError("Malformed signature header")

    mac = hmac.new(GITHUB_SECRET, raw_body, hashlib.sha256)
    expected_sig = mac.hexdigest()

    if not hmac.compare_digest(expected_sig, received_sig):
        raise ValueError("Signature verification failed")
    return True

def trigger_jenkins(payload):
    """
    Call Jenkins with the appropriate parameters.
    """
    data = {
        "token": JENKINS_TOKEN,
        # Example: forward the branch name
        "BRANCH": payload.get("ref", "").split("/")[-1],
    }
    response = requests.post(
        JOB_URL,
        data=data,
        auth=HTTPBasicAuth(JENKINS_USER, JENKINS_TOKEN),
        verify=False,  # set to True with proper certs
        timeout=10,
    )
    response.raise_for_status()
    return response.text

def lambda_handler(event, context):
    try:
        raw_body, headers = extract_payload(event)
        verify_signature(raw_body, headers)

        # Parse JSON *after* verification
        payload = json.loads(raw_body)

        # Optional: filter branches, actions, etc.
        if payload.get("ref") != "refs/heads/main":
            return {"statusCode": 200, "body": "Ignored non‑main branch"}

        trigger_jenkins(payload)

        return {"statusCode": 200, "body": "Jenkins job triggered"}
    except Exception as e:
        # Log the error (CloudWatch) and return a 400/500 as appropriate
        print(f"Error: {e}")
        return {"statusCode": 400, "body": str(e)}

代码要点

• extract_payload 保留 GitHub 签名的原始字节序列（包括对 payload= 表单编码体的处理）。
• verify_signature 使用 hmac.compare_digest 防止时序攻击。
• trigger_jenkins 通过基本认证（或 API Token）在私有 VPC 网络中调用 Jenkins 任务。

回顾

通过让 Jenkins 完全脱离互联网，并让 API Gateway + Lambda 充当守门人，您可以获得一个 安全、可投入生产 的 GitHub 触发构建解决方案。唯一的“坑”是需要在 API Gateway 中启用请求验证器，以防签名头被剥离。设置好后，整个流程即可顺畅运行。

概述

本文档说明了一个可投入生产的 AWS Lambda 函数，该函数充当 GitHub webhook 与 私有 Jenkins 服务器 之间的安全桥梁。

它：

1. 提取原始 webhook 负载。
2. 验证 GitHub 签名。
3. 过滤事件（仅限 main 分支）。
4. 通过内部网络触发 Jenkins 流水线。

验证 GitHub 签名

def verify_signature(raw_body, headers):
    signature = headers.get("x-hub-signature-256")
    if not signature:
        return False

    expected = "sha256=" + hmac.new(
        GITHUB_SECRET,
        raw_body,
        hashlib.sha256
    ).hexdigest()

    # Securely compare the two values to prevent timing attacks
    return hmac.compare_digest(signature, expected)

• 如果缺少 x-hub-signature-256 头部，请求会立即被拒绝。
• 预期的签名是根据负载的 原始字节 重新计算的。
• hmac.compare_digest 提供常量时间比较。

Source:

主 Lambda 处理程序

def lambda_handler(event, context):
    # 1️⃣ 提取并验证 – 在验证之前绝不解析 JSON
    raw_body, headers = extract_payload(event)

    if not verify_signature(raw_body, headers):
        return {"statusCode": 401, "body": "Invalid GitHub signature"}

    # 2️⃣ 现在可以安全地解析 JSON
    payload = json.loads(raw_body.decode("utf-8"))

    # 3️⃣ 仅处理推送到 main 分支的情况
    ref = payload.get("ref", "")
    if not ref.endswith("/main"):
        return {"statusCode": 200, "body": "Not main branch"}

    # 4️⃣ 触发 Jenkins（私有）
    session = requests.Session()
    session.auth = HTTPBasicAuth(JENKINS_USER, JENKINS_TOKEN)

    # 处理 Jenkins CSRF 防护
    crumb = session.get(f"{JENKINS_URL}/crumbIssuer/api/json").json()
    crumb_header = {crumb["crumbRequestField"]: crumb["crumb"]}

    response = session.post(
        JOB_URL,
        headers=crumb_header,
        params={"BRANCH": "main"},
        timeout=10
    )

    return {
        "statusCode": response.status_code,
        "body": "Jenkins pipeline triggered"
    }

• 在进一步处理之前会阻止无效请求，因此 Jenkins 永远不会收到这些请求。
• JSON 负载仅在签名校验通过后才会被解析。
• 对非 main 分支返回 200 响应，使 webhook 保持满意但不执行任何操作。

理解 Lambda 代码

完整的生产就绪 Lambda 代码

import json
import hmac
import hashlib
import base64
import urllib.parse
import requests
from requests.auth import HTTPBasicAuth

# ── Configuration ────────────────────────────────────────────────────────
GITHUB_SECRET = b"demo-github-secret"

JENKINS_URL   = "https://jenkins.example.com"
JENKINS_USER  = "demo-user"
JENKINS_TOKEN = "demo-api-token"

JOB_URL = f"{JENKINS_URL}/job/demo-pipeline/buildWithParameters"
# ────────────────────────────────────────────────────────────────────────

def extract_payload(event):
    """
    Returns:
        raw_body (bytes): the exact bytes that GitHub sent.
        headers (dict): lower‑cased header dict for easy lookup.
    """
    headers = {k.lower(): v for k, v in (event.get("headers") or {}).items()}
    body = event.get("body", "")

    # API Gateway may Base64‑encode the body
    if event.get("isBase64Encoded"):
        raw_body = base64.b64decode(body)
    else:
        raw_body = body.encode("utf-8")

    # GitHub can send payload as `payload=...` (url‑encoded)
    if raw_body.startswith(b"payload="):
        raw_body = urllib.parse.unquote_to_bytes(
            raw_body.decode().replace("payload=", "", 1)
        )

    return raw_body, headers

def verify_signature(raw_body, headers):
    """
    Re‑creates the HMAC‑SHA256 signature and compares it with the header.
    Returns True if the signature matches, False otherwise.
    """
    signature = headers.get("x-hub-signature-256")
    if not signature:
        return False

    expected = "sha256=" + hmac.new(
        GITHUB_SECRET,
        raw_body,
        hashlib.sha256
    ).hexdigest()

    return hmac.compare_digest(signature, expected)

def lambda_handler(event, context):
    raw_body, headers = extract_payload(event)

    if not verify_signature(raw_body, headers):
        return {"statusCode": 401, "body": "Invalid GitHub signature"}

    payload = json.loads(raw_body.decode("utf-8"))

    # Only act on pushes to the main branch
    ref = payload.get("ref", "")
    if not ref.endswith("/main"):
        return {"statusCode": 200, "body": "Not main branch"}

    # Authenticate to Jenkins
    session = requests.Session()
    session.auth = HTTPBasicAuth(JENKINS_USER, JENKINS_TOKEN)

    # Obtain CSRF crumb
    crumb = session.get(f"{JENKINS_URL}/crumbIssuer/api/json").json()
    crumb_header = {crumb["crumbRequestField"]: crumb["crumb"]}

    # Trigger the pipeline
    response = session.post(
        JOB_URL,
        headers=crumb_header,
        params={"BRANCH": "main"},
        timeout=10
    )

    return {
        "statusCode": response.status_code,
        "body": "Jenkins pipeline triggered"
    }