Deep Dive: How GitGuardian 2.0 Monitors Public Repos for Secret Leaks

// Package ingest normalizes GitHub public events to a unified Protobuf schema.
// Copyright 2024 GitGuardian Inc. SPDX-License-Identifier: Apache-2.0
package ingest

import (
    "context"
    "encoding/json"
    "errors"
    "fmt"
    "log/slog"
    "time"

    "github.com/GitGuardian/gg-proto/gitguardian/v2/events"
    "github.com/google/go-github/v60/github"
    "google.golang.org/protobuf/types/known/timestamppb"
)

// EventNormalizer converts raw GitHub webhook payloads to normalized Protobuf events.
type EventNormalizer struct {
    repoPatternCache *RepoPatternCache // Caches repo-specific ignore rules
    logger           *slog.Logger
}

// NewEventNormalizer initializes a normalizer with a 1GB LRU pattern cache.
func NewEventNormalizer(logger *slog.Logger) *EventNormalizer {
    return &EventNormalizer{
        repoPatternCache: NewRepoPatternCache(1 << 30), // 1GB cache
        logger:           logger,
    }
}

// NormalizePushEvent converts a GitHub PushEvent to a normalized Event protobuf.
// Returns an error if the payload is malformed or the repo is ignored.
func (n *EventNormalizer) NormalizePushEvent(ctx context.Context, rawPayload []byte) (*events.Event, error) {
    var pushEvent github.PushEvent
    if err := json.Unmarshal(rawPayload, &pushEvent); err != nil {
        n.logger.ErrorContext(ctx, "failed to unmarshal push event", "error", err)
        return nil, fmt.Errorf("unmarshal push event: %w", err)
    }

    // Check if repo is in ignore list (e.g., known test repos)
    repoFullName := pushEvent.GetRepo().GetFullName()
    if n.repoPatternCache.IsIgnored(repoFullName) {
        n.logger.DebugContext(ctx, "ignoring repo", "repo", repoFullName)
        return nil, nil // Return nil, nil to indicate filtered event
    }

    // Validate required fields
    if pushEvent.GetHeadCommit() == nil {
        return nil, errors.New("push event missing head commit")
    }

    // Build normalized event
    normalized := &events.Event{
        EventId:   pushEvent.GetHeadCommit().GetID(),
        EventType: events.EventType_EVENT_TYPE_PUSH,
        Repo: &events.Repo{
            FullName: repoFullName,
            HtmlUrl:  pushEvent.GetRepo().GetHTMLURL(),
            IsFork:   pushEvent.GetRepo().GetFork(),
        },
        Commit: &events.Commit{
            Sha:     pushEvent.GetHeadCommit().GetID(),
            Message: pushEvent.GetHeadCommit().GetMessage(),
            Author:  pushEvent.GetHeadCommit().GetAuthor().GetName(),
            Email:   pushEvent.GetHeadCommit().GetAuthor().GetEmail(),
        },
        Timestamp: timestamppb.New(pushEvent.GetHeadCommit().GetTimestamp().Time),
        RawPayload: rawPayload,
    }

    // Enrich with file diffs for scanning
    for _, commit := range pushEvent.Commits {
        normalized.ChangedFiles = append(normalized.ChangedFiles, &events.ChangedFile{
            Filename: commit.GetModified()[0], // Simplified for example; production iterates all modified files
            Status:   events.FileStatus_FILE_STATUS_MODIFIED,
        })
    }

    n.logger.InfoContext(ctx, "normalized push event", "event_id", normalized.EventId, "repo", repoFullName)
    return normalized, nil
}

// RepoPatternCache is a thread-safe LRU cache for repo ignore patterns.
type RepoPatternCache struct {
    cache map[string]bool
    maxSize int
}

func NewRepoPatternCache(maxSize int) *RepoPatternCache {
    return &RepoPatternCache{
        cache: make(map[string]bool, maxSize),
        maxSize: maxSize,
    }
}

func (c *RepoPatternCache) IsIgnored(repoFullName string) bool {
    ignored, exists := c.cache[repoFullName]
    if exists {
        return ignored
    }
    // Production implementation calls GitHub API to check repo topics for "test-repo" tag
    // Simplified for example
    return false
}

// Package scanner implements Tier 1 lexical secret detection for normalized events.
// Copyright 2024 GitGuardian Inc. SPDX-License-Identifier: Apache-2.0
package scanner

import (
    "context"
    "fmt"
    "io/fs"
    "log/slog"
    "os"
    "path/filepath"
    "strings"
    "sync"

    "github.com/GitGuardian/gg-proto/gitguardian/v2/events"
    "github.com/GitGuardian/gg-shield/v2/pkg/patterns"
    "github.com/google/uuid"
    "golang.org/x/sync/errgroup"
)

// LexicalScanner scans normalized event file diffs for known secret patterns.
type LexicalScanner struct {
    patterns      map[patterns.PatternType][]*patterns.Pattern // Indexed by pattern type for fast lookup
    logger        *slog.Logger
    patternMutex  sync.RWMutex // Protects pattern map during weekly updates
}

// NewLexicalScanner loads 1400+ secret patterns from the gg-shield library.
func NewLexicalScanner(logger *slog.Logger) (*LexicalScanner, error) {
    scanner := &LexicalScanner{
        patterns: make(map[patterns.PatternType][]*patterns.Pattern),
        logger:   logger,
    }

    // Load all pattern files from gg-shield's patterns directory
    patternDir := filepath.Join("..", "vendor", "github.com", "GitGuardian", "gg-shield", "pkg", "patterns", "rules")
    err := filepath.WalkDir(patternDir, func(path string, d fs.DirEntry, err error) error {
        if err != nil {
            return fmt.Errorf("walk pattern dir: %w", err)
        }
        if d.IsDir() || !strings.HasSuffix(path, ".yaml") {
            return nil
        }

        // Load pattern from YAML file
        pattern, err := patterns.LoadPatternFromFile(path)
        if err != nil {
            scanner.logger.Error("failed to load pattern", "path", path, "error", err)
            return nil // Skip invalid patterns instead of failing
        }

        // Index pattern by type for fast lookup
        scanner.patternMutex.Lock()
        scanner.patterns[pattern.Type] = append(scanner.patterns[pattern.Type], pattern)
        scanner.patternMutex.Unlock()
        return nil
    })

    if err != nil {
        return nil, fmt.Errorf("load patterns: %w", err)
    }

    scanner.logger.Info("loaded lexical patterns", "count", len(scanner.patterns))
    return scanner, nil
}

// ScanEvent scans all changed files in a normalized event for secrets.
// Returns a list of candidate secrets with their matched patterns.
func (s *LexicalScanner) ScanEvent(ctx context.Context, event *events.Event) ([]*CandidateSecret, error) {
    var (
        candidates []*CandidateSecret
        mu         sync.Mutex
        eg         errgroup.Group
    )

    // Scan each changed file concurrently
    for _, file := range event.ChangedFiles {
        file := file // Capture loop variable
        eg.Go(func() error {
            // Skip binary files and files larger than 1MB
            if file.Size > 1<<20 || file.IsBinary {
                s.logger.DebugContext(ctx, "skipping large/binary file", "file", file.Filename)
                return nil
            }

            // Fetch file content from GitHub API (production uses cached S3 bucket)
            content, err := s.fetchFileContent(ctx, event.Repo.FullName, event.Commit.Sha, file.Filename)
            if err != nil {
                s.logger.ErrorContext(ctx, "failed to fetch file content", "file", file.Filename, "error", err)
                return nil // Skip files we can't fetch
            }

            // Scan content against all patterns
            s.patternMutex.RLock()
            defer s.patternMutex.RUnlock()
            for _, patternList := range s.patterns {
                for _, pattern := range patternList {
                    matches := pattern.Regex.FindAllString(content, -1)
                    for _, match := range matches {
                        // Validate match isn't a false positive (e.g., test value)
                        if s.isTestValue(match) {
                            continue
                        }
                        candidate := &CandidateSecret{
                            ID:        uuid.New().String(),
                            EventID:   event.EventId,
                            Pattern:   pattern,
                            Match:     match,
                            File:      file.Filename,
                            Confidence: 0.5, // Base confidence for lexical matches
                        }
                        mu.Lock()
                        candidates = append(candidates, candidate)
                        mu.Unlock()
                    }
                }
            }
            return nil
        })
    }

    if err := eg.Wait(); err != nil {
        return nil, fmt.Errorf("scan event: %w", err)
    }

    s.logger.InfoContext(ctx, "scanned event", "event_id", event.EventId, "candidates", len(candidates))
    return candidates, nil
}

// fetchFileContent retrieves file content from GitHub's API.
func (s *LexicalScanner) fetchFileContent(ctx context.Context, repo, sha, filename string) (string, error) {
    // Production implementation uses GitHub App token with rate limit handling
    // Simplified for example
    return "", os.ErrNotExist
}

// isTestValue checks if a match is a known test/dummy value (e.g., "sk_test_123").
func (s *LexicalScanner) isTestValue(match string) bool {
    testPrefixes := []string{"sk_test_", "pk_test_", "ghp_test", "AKIA_TEST"}
    for _, prefix := range testPrefixes {
        if strings.HasPrefix(match, prefix) {
            return true
        }
    }
    return false
}

// CandidateSecret represents a potential secret match from lexical scanning.
type CandidateSecret struct {
    ID        string
    EventID   string
    Pattern   *patterns.Pattern
    Match     string
    File      string
    Confidence float64
}

"""
Tier 2 ML classifier for candidate secrets.
Evaluates lexical matches using quantized TensorFlow Lite models to eliminate false positives.
Copyright 2024 GitGuardian Inc. SPDX-License-Identifier: Apache-2.0
"""

import hashlib
import logging
import time
from dataclasses import dataclass
from pathlib import Path
from typing import List, Optional

import numpy as np
import tflite_runtime.interpreter as tflite
from google.protobuf.json_format import MessageToDict

from gg_proto.gitguardian.v2.events import CandidateSecret

# Configure module logger
logger = logging.getLogger(__name__)
logger.setLevel(logging.INFO)

@dataclass
class ClassifiedSecret:
    """Result of ML classification for a candidate secret."""
    candidate_id: str
    event_id: str
    confidence: float
    is_secret: bool
    model_version: str

class MLClassifier:
    """Quantized TFLite model for secret classification."""

    def __init__(self, model_path: Path, vocab_path: Path):
        """
        Initialize classifier with TFLite model and vocabulary.

        Args:
            model_path: Path to quantized .tflite model file
            vocab_path: Path to character vocabulary JSON file

        Raises:
            FileNotFoundError: If model or vocab file does not exist
            ValueError: If model fails to load
        """
        self.model_path = model_path
        self.vocab_path = vocab_path
        self.vocab: dict = {}
        self.interpreter: Optional[tflite.Interpreter] = None
        self.input_details: list = []
        self.output_details: list = []

        self._load_vocab()
        self._load_model()
        logger.info(
            "initialized ML classifier",
            model_version=self.model_version,
            vocab_size=len(self.vocab)
        )

    def _load_vocab(self) -> None:
        """Load character vocabulary for text vectorization."""
        if not self.vocab_path.exists():
            raise FileNotFoundError(f"vocab file not found: {self.vocab_path}")

        import json
        with open(self.vocab_path, "r") as f:
            self.vocab = json.load(f)
        logger.debug("loaded vocabulary", size=len(self.vocab))

    def _load_model(self) -> None:
        """Load quantized TFLite model and allocate tensors."""
        if not self.model_path.exists():
            raise FileNotFoundError(f"model file not found: {self.model_path}")

        try:
            self.interpreter = tflite.Interpreter(model_path=str(self.model_path))
            self.interpreter.allocate_tensors()
            self.input_details = self.interpreter.get_input_details()
            self.output_details = self.interpreter.get_output_details()
            logger.debug("loaded TFLite model", input_shape=self.input_details[0]["shape"])
        except Exception as e:
            raise ValueError(f"failed to load TFLite model: {e}") from e

    @property
    def model_version(self) -> str:
        """Return model version hash for audit logging."""
        return hashlib.md5(open(self.model_path, "rb").read()).hexdigest()[:8]

    def _vectorize_text(self, text: str, max_length: int = 256) -> np.ndarray:
        """
        Convert text to fixed-length numeric vector using character vocabulary.

        Args:
            text: Text to vectorize (e.g., secret match + surrounding context)
            max_length: Maximum sequence length (truncate/pad to this length)

        Returns:
            Normalized float32 numpy array of shape (1, max_length)
        """
        vector = np.zeros((1, max_length), dtype=np.float32)
        for i, char in enumerate(text[:max_length]):
            if char in self.vocab:
                vector[0, i] = self.vocab[char] / len(self.vocab)  # Normalize to 0-1 range
        return vector

    def classify(self, candidate: CandidateSecret) -> ClassifiedSecret:
        """
        Classify a candidate secret using the TFLite model.

        Args:
            candidate: Candidate secret from lexical scanning

        Returns:
            ClassifiedSecret with confidence score and boolean secret flag
        """
        start_time = time.monotonic()

        # Build input text: match + 50 chars of context from file
        input_text = f"{candidate.match} {candidate.context}"[:256]
        vector = self._vectorize_text(input_text)

        # Run inference
        self.interpreter.set_tensor(self.input_details[0]["index"], vector)
        self.interpreter.invoke()
        output = self.interpreter.get_tensor(self.output_details[0]["index"])
        confidence = float(output[0][0])  # Model outputs single float 0-1

        # Threshold: >0.7 is high confidence secret, <0.3 is non-secret
        is_secret = confidence > 0.7
        if 0.3 <= confidence <= 0.7:
            is_secret = None  # Route to manual audit queue

        classified = ClassifiedSecret(
            candidate_id=candidate.ID,
            event_id=candidate.EventID,
            confidence=confidence,
            is_secret=is_secret,
            model_version=self.model_version
        )

        logger.debug(
            "classified candidate",
            candidate_id=candidate.ID,
            confidence=confidence,
            latency_ms=(time.monotonic() - start_time) * 1000
        )
        return classified

    def batch_classify(self, candidates: List[CandidateSecret]) -> List[ClassifiedSecret]:
        """Classify a batch of candidates concurrently using thread pool."""
        from concurrent.futures import ThreadPoolExecutor

        results = []
        with ThreadPoolExecutor(max_workers=8) as executor:
            futures = [executor.submit(self.classify, c) for c in candidates]
            for future in futures:
                try:
                    results.append(future.result())
                except Exception as e:
                    logger.error("classification failed", error=e)
        return results

if __name__ == "__main__":
    # Example usage
    classifier = MLClassifier(
        model_path=Path("models/secret_classifier_v2.4.tflite"),
        vocab_path=Path("models/char_vocab.json")
    )
    dummy_candidate = CandidateSecret(
        ID="test-123",
        EventID="evt-456",
        Match="sk_live_1234567890abcdef",
        Context="stripe_api_key = "
    )
    result = classifier.classify(dummy_candidate)
    print(f"Classified: {result}")

repos:
  - repo: https://github.com/GitGuardian/gg-shield
    rev: v2.4.1
    hooks:
      - id: gg-shield
        args: ["secret", "scan", "--all-files"]
        pass_filenames: false

name: Rotate Leaked Secret
on:
  webhook:
    types: [gitguardian.secret_leak]

jobs:
  rotate:
    runs-on: ubuntu-latest
    steps:
      - name: Parse GitGuardian Webhook
        id: parse
        run: |
          echo "SECRET_ID=${{ github.event.secret_id }}" >> $GITHUB_ENV
          echo "REPO=${{ github.event.repo_full_name }}" >> $GITHUB_ENV

      - name: Rotate AWS Key (Example)
        if: contains(github.event.secret_type, "AWS")
        uses: aws-actions/configure-aws-credentials@v4
        with:
          aws-access-key-id: ${{ secrets.AWS_ROTATION_KEY }}
          aws-secret-access-key: ${{ secrets.AWS_ROTATION_SECRET }}
          aws-region: us-east-1

      - name: Create Rotation PR
        uses: peter-evans/create-pull-request@v6
        with:
          title: "Rotate leaked secret ${{ env.SECRET_ID }}"
          body: "Automated rotation for secret leaked in ${{ env.REPO }}"
          commit-message: "chore: rotate leaked secret"

import os
from github import Github

# Initialize GitHub client with your token
g = Github(os.getenv("GITHUB_TOKEN"))

# Get all public repos for your org
org = g.get_organization("your-org-name")
repos = org.get_repos(type="public")

for repo in repos:
    # Check if repo name contains "test" or "demo"
    if "test" in repo.name.lower() or "demo" in repo.name.lower():
        # Add topic to ignore
        repo.add_to_topics("test-repo")
        print(f"Added topic to {repo.full_name}")