Security: Implement GitHub Webhook Signature Verification and Idempotency Guard

[Shevilll]

Security: Implement GitHub Webhook Signature Verification and Idempotency Guard

Context & Importance

As part of the backend development and integration of GitHub webhooks for AOSSIE's InPactAI (following up on the requirements discussed in #293), it is critical to protect the platform's API endpoints from malicious or unintended actions. Since InPactAI manages open-source contribution flows, active task tracking, and platform-wide assignment states, our webhook handlers represent a critical attack surface.

Without robust validation layers, the application is highly vulnerable to:

1. Spam Event Injections: Attackers can forge webhooks, spoofing GitHub's payload format to arbitrarily assign issues/PRs or alter contributor statuses without authentication.
2. Replay Attacks (Duplicate Events): Network delays, timeouts, or intentional malicious delivery replays can trigger duplicate handlers, leading to database state corruption and invalid active contributor stats.
3. Over-Assignment / Spam Claiming: Bad actors or spam bots could exploit the platform by claiming multiple concurrent assignments to block genuine contributors.

Implementing these webhook security guards ensures complete integrity of contributor tracking and keeps the platform secure, robust, and idempotent.

---

1. System Architecture & Event Flow

The sequence below illustrates the lifecycle of a secure, authenticated, and deduplicated webhook request as it travels from the GitHub API through our FastAPI application down to the database:

sequenceDiagram
    autonumber
    participant GitHub as GitHub API
    participant Middleware as security_guard (FastAPI)
    participant DB as Database (Postgres)
    participant Handler as Webhook Service

    GitHub->>Middleware: POST /routes/github/webhook (with X-Hub-Signature-256)
    
    rect rgb(240, 248, 255)
        note right of Middleware: Step 1: Authentication Guard
        Middleware->>Middleware: Verify HMAC SHA-256 Signature
        alt Invalid Signature
            Middleware-->>GitHub: 403 Forbidden (Aborted)
        end
    end

    rect rgb(245, 245, 245)
        note right of Middleware: Step 2: Replay/Deduplication Guard
        Middleware->>DB: Query X-GitHub-Delivery ID
        alt Already Processed
            DB-->>Middleware: Duplicate Detected
            Middleware-->>GitHub: 200 OK (Ignored/No-op)
        end
        Middleware->>DB: Save X-GitHub-Delivery ID as 'pending'
    end

    rect rgb(255, 245, 238)
        note right of Handler: Step 3: Assignment & Spam Guard
        Middleware->>Handler: Forward Validated Payload
        Handler->>Handler: Validate Repo Whitelist & Collaborator Permissions
        Handler->>Handler: Check Assignee Concurrent Assignment Limits
        alt Spam / Over-assignment Detected
            Handler-->>GitHub: 400 Bad Request / 200 OK (Log Spam Warning)
        end
    end

    Handler->>DB: Update Issue/PR Status & Record Assignment
    Handler-->>GitHub: 200 OK (Success)

Loading

---

2. Webhook Authentication Guard (HMAC SHA-256)

GitHub signs each webhook payload with a cryptographic hash using a pre-configured secret. We must calculate the signature of the raw incoming request body and match it with the X-Hub-Signature-256 header.

Security Best Practices

• Raw Body Requirement: Always use the raw byte body to compute the HMAC signature; parsing and re-serializing JSON may alter whitespace, breaking validation.
• Timing-Safe Comparison: Use hmac.compare_digest to compare signatures, preventing side-channel timing attacks.
• Secret Configuration: Store the webhook secret securely using environment variables (GITHUB_WEBHOOK_SECRET).

Python / FastAPI Implementation

import hmac
import hashlib
import os
from fastapi import Request, HTTPException, Security, status

GITHUB_WEBHOOK_SECRET = os.getenv("GITHUB_WEBHOOK_SECRET")

async def verify_github_signature(request: Request):
    """
    FastAPI Dependency to verify the incoming X-Hub-Signature-256 header.
    Raises 403 Forbidden for unauthorized requests.
    """
    if not GITHUB_WEBHOOK_SECRET:
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail="Server misconfiguration: GitHub Webhook Secret is missing."
        )

    signature_header = request.headers.get("X-Hub-Signature-256")
    if not signature_header:
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="Missing X-Hub-Signature-256 header."
        )

    # Header is in the format "sha256=HEX_SIGNATURE"
    parts = signature_header.split("=")
    if len(parts) != 2 or parts[0] != "sha256":
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="Invalid signature header format. Only sha256 is supported."
        )
    signature_to_verify = parts[1]

    # Get the raw request payload
    raw_body = await request.body()

    # Calculate expected signature
    computed_signature = hmac.new(
        key=GITHUB_WEBHOOK_SECRET.encode("utf-8"),
        msg=raw_body,
        digestmod=hashlib.sha256
    ).hexdigest()

    # Timing-safe string comparison
    if not hmac.compare_digest(computed_signature, signature_to_verify):
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="Signature validation failed. Unauthenticated source."
        )

---

3. Idempotency Guard (Deduplication)

GitHub may occasionally retry a webhook request (e.g., due to network timeouts, connection drops, or cold startup delays in the backend), leading to duplicate issue/PR processings.

To guarantee exactly-once processing, we implement an Idempotency Guard:

1. Every payload includes a unique delivery GUID in the X-GitHub-Delivery header.
2. We query the database to see if X-GitHub-Delivery exists in our processed_webhooks table.
3. If it exists, we immediately terminate processing with 200 OK (to prevent GitHub from marking it as failed).
4. If it is new, we store the UUID with a status of processing and proceed.

---

4. Assignment Validation & Spam Prevention Logic

Once authenticated and deduplicated, we validate the event metadata to reject bad actors, spam bots, and unauthorized changes.

A. Repository Whitelisting

To prevent cross-repository injection attacks, we only accept events from repositories that are explicitly registered on the InPactAI platform:

• Match payload["repository"]["id"] or payload["repository"]["full_name"] against our registered repository configurations.

B. Assignment Authority Guard

In a collaborative open-source platform, users shouldn't be allowed to maliciously trigger assignment webhooks using fake setups or mock events.

• Check the Actor: The payload["sender"]["login"] represents the actor who performed the assignment.
• Authorization Check: Query the GitHub API (or check a cached registry of repository collaborators) to verify if the actor has admin, write, or authorized maintainer privileges on that repository.

C. Contributor Concurrency Limit (Spam Claiming)

A common spam pattern in GSoC/hackathons is a single user claiming 5+ issues at once to block others, then abandoning them.

• Limit Active Assignments: Restrict a single contributor to a maximum of 2 active assigned issues/PRs across the platform.
• Enforcement: If a webhook reports a new assignment for a contributor who already has active assignments over the threshold, either:
  1. Comment automatically on the issue using a bot to unassign them.
  2. Reject the sync internally and flag a warning.

D. Event Context Validation

Ensure that the issue state matches the event action:

• For issues.assigned events, verify that the issue is currently open.
• For pull_request events, reject assignment attempts if the PR is already closed or merged.

---

5. SQLAlchemy Database Schemas

To support these security guards, we define three tables:

1. processed_webhooks: Stores delivery IDs for idempotency checks.
2. registered_repositories: Stores platform-approved repositories and their optional per-repo credentials.
3. active_assignments: Tracks active work to enforce concurrency and claim integrity.

from sqlalchemy import Column, String, Integer, BigInteger, DateTime, ForeignKey, Boolean, UniqueConstraint
from sqlalchemy.orm import relationship
from datetime import datetime
from app.db.db import Base

class ProcessedWebhook(Base):
    """
    Idempotency Table to track processed webhook deliveries and prevent replay attacks.
    Deliveries should be pruned periodically (e.g., via cron) after 7 days.
    """
    __tablename__ = "processed_webhooks"

    delivery_id = Column(String(50), primary_key=True, index=True) # UUID format from X-GitHub-Delivery
    event_type = Column(String(50), nullable=False)                # from X-GitHub-Event
    status = Column(String(20), default="processing")              # processing, completed, failed
    processed_at = Column(DateTime, default=datetime.utcnow, index=True)

class RegisteredRepository(Base):
    """
    White-list of repositories integrated with the InPactAI platform.
    """
    __tablename__ = "registered_repositories"

    id = Column(BigInteger, primary_key=True)                      # GitHub's repository ID
    full_name = Column(String(255), unique=True, nullable=False)   # owner/repo-name
    is_active = Column(Boolean, default=True)
    created_at = Column(DateTime, default=datetime.utcnow)

    assignments = relationship("ActiveAssignment", back_populates="repository")

class ActiveAssignment(Base):
    """
    Table to track active contributor assignments.
    Prevents spam over-claiming and provides state tracking.
    """
    __tablename__ = "active_assignments"

    id = Column(String, primary_key=True)                          # Platform custom assignment UUID
    repository_id = Column(BigInteger, ForeignKey("registered_repositories.id"), nullable=False)
    issue_number = Column(Integer, nullable=False)                 # Issue/PR number
    target_type = Column(String(10), nullable=False)               # 'issue' or 'pull_request'
    
    assignee_github_id = Column(BigInteger, nullable=False, index=True)
    assignee_username = Column(String(100), nullable=False)
    
    assigned_by_github_id = Column(BigInteger, nullable=False)     # The moderator/bot who performed assignment
    assigned_at = Column(DateTime, default=datetime.utcnow)
    status = Column(String(20), default="active", index=True)      # active, completed, abandoned
    
    repository = relationship("RegisteredRepository", back_populates="assignments")

    __table_args__ = (
        UniqueConstraint('repository_id', 'issue_number', 'assignee_github_id', name='_repo_issue_assignee_uc'),
    )

---

6. End-to-End FastAPI Router Blueprint

Integrating all security layers together, here is the robust router blueprint for /routes/github:

from fastapi import APIRouter, Request, Depends, status, Response
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.future import select
from app.db.db import get_db
from .security import verify_github_signature
from ..models import ProcessedWebhook, RegisteredRepository, ActiveAssignment
from datetime import datetime

router = APIRouter(prefix="/github", tags=["GitHub Integration"])

@router.post(
    "/webhook",
    dependencies=[Depends(verify_github_signature)],
    status_code=status.HTTP_200_OK
)
async def handle_github_webhook(
    request: Request,
    db: AsyncSession = Depends(get_db)
):
    # 1. Fetch Webhook Metadata headers
    delivery_id = request.headers.get("X-GitHub-Delivery")
    event_type = request.headers.get("X-GitHub-Event")

    if not delivery_id:
        return Response(status_code=status.HTTP_400_BAD_REQUEST, content="Missing Delivery ID Header.")

    # 2. Idempotency Check (Deduplication)
    stmt = select(ProcessedWebhook).where(ProcessedWebhook.delivery_id == delivery_id)
    result = await db.execute(stmt)
    existing_delivery = result.scalars().first()

    if existing_delivery:
        # Deliveries already successfully recorded return 200 immediately to avoid hook timeout retries
        return {"status": "ignored", "reason": f"Event {delivery_id} already processed."}

    # Record the delivery to lock processing
    webhook_log = ProcessedWebhook(delivery_id=delivery_id, event_type=event_type, status="processing")
    db.add(webhook_log)
    await db.commit()

    # 3. Parse JSON Body
    payload = await request.json()

    try:
        # 4. Whitelist Validation
        repo_id = payload.get("repository", {}).get("id")
        repo_stmt = select(RegisteredRepository).where(RegisteredRepository.id == repo_id)
        repo_result = await db.execute(repo_stmt)
        repo = repo_result.scalars().first()

        if not repo or not repo.is_active:
            webhook_log.status = "ignored"
            await db.commit()
            return {"status": "ignored", "reason": "Repository is not registered or is inactive."}

        # 5. Route Specific Event Handlers
        if event_type == "issues":
            action = payload.get("action")
            if action in ["assigned", "unassigned"]:
                await process_issue_assignment(payload, db)

        elif event_type == "pull_request":
            action = payload.get("action")
            if action in ["assigned", "unassigned", "opened", "closed"]:
                await process_pr_assignment(payload, db)

        # Mark log as completed
        webhook_log.status = "completed"
        await db.commit()
        return {"status": "success", "delivery_id": delivery_id}

    except Exception as e:
        # Mark log as failed to allow debug visibility
        webhook_log.status = "failed"
        await db.commit()
        return {"status": "error", "message": str(e)}


async def process_issue_assignment(payload: dict, db: AsyncSession):
    """
    Validates assignment details, checks active claim thresholds and handles storage.
    """
    action = payload.get("action")
    issue = payload.get("issue", {})
    assignee = payload.get("assignee", {})
    sender = payload.get("sender", {})
    repo_id = payload.get("repository", {}).get("id")
    issue_num = issue.get("number")

    if action == "assigned":
        # Check active assignment count for this contributor to prevent claiming spam
        stmt = select(ActiveAssignment).where(
            ActiveAssignment.assignee_github_id == assignee.get("id"),
            ActiveAssignment.status == "active"
        )
        res = await db.execute(stmt)
        active_assignments = res.scalars().all()

        # Enforce Concurrency Threshold
        MAX_CONCURRENT_ASSIGNMENTS = 2
        if len(active_assignments) >= MAX_CONCURRENT_ASSIGNMENTS:
            # OPTIONAL: trigger GitHub Comment API / dispatch bot action to post warning on the issue thread
            # logger.warning(f"User {assignee.get('login')} exceeded concurrency threshold limit.")
            pass

        # Save assignment
        new_assignment = ActiveAssignment(
            id=f"assign-{repo_id}-{issue_num}-{assignee.get('id')}",
            repository_id=repo_id,
            issue_number=issue_num,
            target_type="issue",
            assignee_github_id=assignee.get("id"),
            assignee_username=assignee.get("login"),
            assigned_by_github_id=sender.get("id"),
            status="active"
        )
        db.add(new_assignment)
        await db.commit()

    elif action == "unassigned":
        # Resolve active assignment
        stmt = select(ActiveAssignment).where(
            ActiveAssignment.repository_id == repo_id,
            ActiveAssignment.issue_number == issue_num,
            ActiveAssignment.assignee_github_id == assignee.get("id"),
            ActiveAssignment.status == "active"
        )
        res = await db.execute(stmt)
        assignment = res.scalars().first()
        if assignment:
            assignment.status = "abandoned"
            await db.commit()


async def process_pr_assignment(payload: dict, db: AsyncSession):
    # Similar validation flow mapping PR hooks to active assignments table.
    pass