GitHub Actions 기반 저장소 모니터링 및 알림 시스템 구축

보안 커뮤니티에서 GitHub 저장소를 지속적으로 감시하여 새로운 도구와 취약점 정보를 수집하는 워크플로우는 이미 여러 프로젝트에서 구현되었다. 대표적인 예로 코드 누수 탐지 도구들이 있으며, 이러한 솔루션들을 확장하여 특정 키워드 기반의 저장소 모니터링 시스템을 구축할 수 있다.

핵심 과제는 지속적인 실행 환경을 유지하는 것이다. 전통적인 방식은 VPS에서 스크립트를 백그라운드로 실행하는 것이지만, GitHub Actions를 활용하면 인프라 비용 없이 동일한 기능을 구현할 수 있다.

GitHub Actions 핵심 개념

GitHub Actions는 저장소 내에서 자동화된 워크플로우를 실행하는 플랫폼이다. 주요 구성 요소는 다음과 같다:

  • Workflow: 전체 자동화 프로세스의 정의
  • Job: 워크플로우 내 병렬 또는 순차 실행 가능한 작업 단위
  • Step: Job을 구성하는 개별 명령 집합
  • Action: 재사용 가능한 특정 작업 단위

기본 워크플로우 구성

최소한의 워크플로우는 다음과 같이 정의된다:

name: Repository Monitor

on:
  push:
    branches: [main]
  workflow_dispatch:

jobs:
  monitor:
    runs-on: ubuntu-latest
    steps:
      - name: Checkout source
        uses: actions/checkout@v4
      
      - name: Execute monitoring
        run: echo "Monitoring started"

스케줄 기반 실행 설정

정기적 실행을 위해 schedule 이벤트를 사용한다. Cron 표현식은 5분 간격이 최소 단위이며, GitHub의 로드 밸런싱으로 인해 실제 실행 시점은 지정 시간 ±10~30분 범위에서 변동될 수 있다.

on:
  schedule:
    - cron: '0 4 * * *'  # UTC 04:00, KST 13:00

한국 시간(KST) 적용을 위해 환경 변수를 설정한다:

env:
  TZ: Asia/Seoul

Python 모니터링 스크립트 구현

GitHub API를 활용하여 저장소를 검색하고, 결과를 메신저로 전송하는 로직을 구현한다.

핵심 로직 구조

import os
import sys
import requests
import datetime
import re
from urllib.parse import quote

class RepoMonitor:
    def __init__(self):
        self.api_token = os.environ.get("GH_ACCESS_TOKEN")
        self.search_terms = self._parse_terms()
        self.seen_repos = self._load_history()
        self.headers = {
            "Authorization": f"token {self.api_token}",
            "Accept": "application/vnd.github.v3+json"
        }
    
    def _parse_terms(self):
        raw = os.environ.get("SEARCH_TERMS", "")
        return [t.strip() for t in raw.split() if t.strip()]
    
    def _load_history(self):
        history = os.environ.get("PROCESSED_REPOS", "")
        return set(history.split()) if history else set()
    
    def fetch_repositories(self, term):
        encoded = quote(term)
        endpoint = f"https://api.github.com/search/repositories?q={encoded}&sort=updated&order=desc"
        
        try:
            resp = requests.get(endpoint, headers=self.headers, timeout=15)
            resp.raise_for_status()
            return resp.json().get("items", [])
        except Exception as err:
            print(f"API 호출 실패: {err}")
            return []
    
    def filter_daily_updates(self, items):
        today = datetime.date.today().isoformat()
        fresh = []
        
        for idx, repo in enumerate(items[:15]):
            pushed = repo.get("pushed_at", "")
            date_match = re.search(r'\d{4}-\d{2}-\d{2}', pushed)
            
            if date_match and date_match.group() == today:
                identifier = repo.get("full_name", repo.get("name"))
                if identifier not in self.seen_repos:
                    fresh.append({
                        "name": identifier,
                        "url": repo.get("html_url"),
                        "desc": repo.get("description") or "설명 없음",
                        "lang": repo.get("language") or "미지정"
                    })
                    self.seen_repos.add(identifier)
        
        return fresh
    
    def execute(self):
        notifications = []
        
        for term in self.search_terms:
            results = self.fetch_repositories(term)
            new_items = self.filter_daily_updates(results)
            notifications.extend(new_items)
        
        if notifications:
            self._send_aggregate(notifications)
            self._persist_state()
        
        return len(notifications)
    
    def _send_aggregate(self, data):
        # 메신저 전송 로직 (구현 예시)
        pass
    
    def _persist_state(self):
        # 상태 저장 로직
        pass

if __name__ == "__main__":
    monitor = RepoMonitor()
    count = monitor.execute()
    print(f"처리 완료: {count}개 저장소")
    sys.exit(0)

상태 영속화 전략

GitHub Actions의 실행 환경은 매번 초기화되므로, 이미 처리한 저장소 정보를 외부에 보관해야 한다. 저장소 내 파일을 활용하는 방식이 가장 간단하다.

워크플로우에서 상태 파일을 환경 변수로 로드:

- name: Load processed history
  run: |
    if [ -f ./processed.db ]; then
      echo "PROCESSED_REPOS=$(cat ./processed.db | tr '\n' ' ')" >> $GITHUB_ENV
    else
      echo "PROCESSED_REPOS=" >> $GITHUB_ENV
    fi

Python에서 상태 갱신 및 환경 변수 반영:

def _persist_state(self):
    serialized = " ".join(self.seen_repos)
    env_path = os.getenv("GITHUB_ENV")
    
    if env_path:
        with open(env_path, "a") as f:
            f.write(f"\nPROCESSED_REPOS={serialized}\n")
    
    # 로컬 파일에도 백업
    with open("./processed.db", "w") as f:
        f.write("\n".join(self.seen_repos))

워크플로우 완성본

name: Daily Repo Monitor

on:
  schedule:
    - cron: '0 4 * * *'
  workflow_dispatch:

env:
  TZ: Asia/Seoul

jobs:
  scan:
    runs-on: ubuntu-latest
    steps:
      - name: Checkout repository
        uses: actions/checkout@v4
      
      - name: Restore history
        run: |
          touch processed.db
          echo "PROCESSED_REPOS=$(cat processed.db | tr '\n' ' ')" >> $GITHUB_ENV
      
      - name: Setup Python
        uses: actions/setup-python@v5
        with:
          python-version: '3.11'
      
      - name: Install dependencies
        run: |
          pip install --upgrade pip
          pip install requests
      
      - name: Execute monitor
        env:
          GH_ACCESS_TOKEN: ${{ secrets.GITHUB_TOKEN }}
          SEARCH_TERMS: ${{ vars.MONITOR_KEYWORDS }}
          WEBHOOK_URL: ${{ secrets.NOTIFY_ENDPOINT }}
          SIGN_SECRET: ${{ secrets.NOTIFY_SECRET }}
        run: python monitor.py
      
      - name: Commit history
        run: |
          git config user.name "github-actions[bot]"
          git config user.email "actions@github.com"
          git add processed.db
          git diff --cached --quiet || git commit -m "Update: $(date +%Y-%m-%d)"
          git push

알림 채널 통합

다양한 메신저 플랫폼으로 결과를 전송할 수 있다. 다음은 일반적인 HTTP 웹훅 기반 전송 예시:

import hmac
import hashlib
import base64
import json

class Notifier:
    def __init__(self, webhook, secret=None):
        self.webhook = webhook
        self.secret = secret
    
    def _generate_signature(self, payload):
        if not self.secret:
            return None
        timestamp = str(int(datetime.datetime.now().timestamp() * 1000))
        string_to_sign = f"{timestamp}\n{self.secret}"
        sign = hmac.new(
            self.secret.encode('utf-8'),
            string_to_sign.encode('utf-8'),
            hashlib.sha256
        ).digest()
        return timestamp, base64.b64encode(sign).decode('utf-8')
    
    def send(self, title, entries):
        if not entries:
            return
        
        blocks = []
        for entry in entries[:8]:  # 최대 8개 제한
            blocks.append(f"**{entry['name']}**\n{entry['desc'][:60]}...\n🔗 {entry['url']}")
        
        message = {
            "title": f"{title} ({len(entries)}개 감지)",
            "content": "\n\n".join(blocks),
            "timestamp": datetime.datetime.now().isoformat()
        }
        
        headers = {"Content-Type": "application/json"}
        if self.secret:
            ts, sig = self._generate_signature(json.dumps(message))
            headers["X-Timestamp"] = ts
            headers["X-Signature"] = sig
        
        requests.post(self.webhook, json=message, headers=headers, timeout=10)

보안 고려사항

민감한 설정값은 반드시 GitHub Secrets/Variables에 저장해야 한다:

설정 위치용도예시
Repository SecretsAPI 토큰, 웹훈 시크릿NOTIFY_SECRET
Repository Variables검색 키워드, 설정값MONITOR_KEYWORDS

워크플로우에서 참조:

env:
  SEARCH_TERMS: ${{ vars.MONITOR_KEYWORDS }}
  WEBHOOK_SECRET: ${{ secrets.NOTIFY_SECRET }}

확장 가능한 기능

  • 다중 소스 통합: GitLab, Bitbucket API 추가
  • 스마트 필터링: 저장소 별점, 포크 수 기준 필터
  • 분류 태깅: 언어별, 주제별 자동 분류
  • 중복 감지: 내용 해시 기반 중복 제거
  • 쓰로틀링: 시간대별 전송량 제한

트러블슈팅

일반적인 문제와 해결 방안:

API 속도 제한
인증된 요청으로 분당 30회, 시간당 5,000회 한도. 검색어를 분산하여 실행 시간을 조정
빈 결과 반환
GitHub Search API는 최근 1년 내 업데이트된 저장소만 인덱싱. created: 한정자 제거 확인
상태 손실
processed.db 파일이 커밋되지 않은 경우. 워크플로우 마지막 단계의 권한과 파일 경로 확인

태그: GitHub Actions GitHub API Python automation Webhook Integration CI/CD Workflow

8월 12일 21:02에 게시됨

인기 태그

©2026 괴물 클럽